Specification
- Skill ID
- affaan-m/ecc/python-patterns
- Publisher
- affaan-m
- Repository
- ecc
- Installs
- 0
- Files
- 1
- Synced
- Sep 23, 2026
How to use it
Open any RiverX project, open the Skills panel in the chat, and search for this identifier. The files are fetched from the source repository at install time.
affaan-m/ecc/python-patternsInstalls these files- SKILL.md
What this skill tells the agent
Python Patterns
This skill provides comprehensive Python patterns extending common design principles with Python-specific idioms.
Protocol (Duck Typing)
Use Protocol for structural subtyping (duck typing with type hints):
from typing import Protocol
class Repository(Protocol):
def find_by_id(self, id: str) -> dict | None: ...
def save(self, entity: dict) -> dict: ...
# Any class with these methods satisfies the protocol
class UserRepository:
def find_by_id(self, id: str) -> dict | None:
# implementation
pass
def save(self, entity: dict) -> dict:
# implementation
pass
def process_entity(repo: Repository, id: str) -> None:
entity = repo.find_by_id(id)
# ... processBenefits:
- Type safety without inheritance
- Flexible, loosely coupled code
- Easy testing and mocking
Dataclasses as DTOs
Use dataclass for data transfer objects and value objects:
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class CreateUserRequest:
name: str
email: str
age: Optional[int] = None
tags: list[str] = field(default_factory=list)
@dataclass(frozen=True)
class User:
"""Immutable user entity"""
id: str
name: str
email: strFeatures:
- Auto-generated
__init__,__repr__,__eq__ frozen=Truefor immutabilityfield()for complex defaults- Type hints for validation
Context Managers
Use context managers (with statement) for resource management:
from contextlib import contextmanager
from typing import Generator
@contextmanager
def database_transaction(db) -> Generator[None, None, None]:
"""Context manager for database transactions"""
try:
yield
db.commit()
except Exception:
db.rollback()
raise
# Usage
with database_transaction(db):
db.execute("INSERT INTO users ...")Class-based context manager:
class FileProcessor:
def __init__(self, filename: str):
self.filename = filename
self.file = None
def __enter__(self):
self.file = open(self.filename, 'r')
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
return False # Don't suppress exceptionsGenerators
Use generators for lazy evaluation and memory-efficient iteration:
def read_large_file(filename: str):
"""Generator for reading large files line by line"""
with open(filename, 'r') as f:
for line in f:
yield line.strip()
# Memory-efficient processing
for line in read_large_file('huge.txt'):
process(line)Generator expressions:
# Instead of list comprehension
squares = (x**2 for x in range(1000000)) # Lazy evaluation
# Pipeline pattern
numbers = (x for x in range(100))
evens = (x for x in numbers if x % 2 == 0)
squares = (x**2 for x in evens)Decorators
Function Decorators
from functools import wraps
import time
def timing(func):
"""Decorator to measure execution time"""
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
end = time.time()
print(f"{func.__name__} took {end - start:.2f}s")
return result
return wrapper
@timing
def slow_function():
time.sleep(1)Class Decorators
def singleton(cls):
"""Decorator to make a class a singleton"""
instances = {}
@wraps(cls)
def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
@singleton
class Config:
passAsync/Await
Async Functions
import asyncio
from typing import List
async def fetch_user(user_id: str) -> dict:
"""Async function for I/O-bound operations"""
await asyncio.sleep(0.1) # Simulate network call
return {"id": user_id, "name": "Alice"}
async def fetch_all_users(user_ids: List[str]) -> List[dict]:
"""Concurrent execution with asyncio.gather"""
tasks = [fetch_user(uid) for uid in user_ids]
return await asyncio.gather(*tasks)
# Run async code
asyncio.run(fetch_all_users(["1", "2", "3"]))Async Context Managers
class AsyncDatabase:
async def __aenter__(self):
await self.connect()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.disconnect()
async with AsyncDatabase() as db:
await db.query("SELECT * FROM users")Type Hints
Advanced Type Hints
from typing import TypeVar, Generic, Callable, ParamSpec, Concatenate
T = TypeVar('T')
P = ParamSpec('P')
class Repository(Generic[T]):
"""Generic repository pattern"""
def __init__(self, entity_type: type[T]):
self.entity_type = entity_type
def find_by_id(self, id: str) -> T | None:
# implementation
pass
# Type-safe decorator
def log_call(func: Callable[P, T]) -> Callable[P, T]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
print(f"Calling {func.__name__}")
return func(*args, **kwargs)
return wrapperUnion Types (Python 3.10+)
def process(value: str | int | None) -> str:
match value:
case str():
return value.upper()
case int():
return str(value)
case None:
return "empty"Dependency Injection
Constructor Injection
class UserService:
def __init__(
self,
repository: Repository,
logger: Logger,
cache: Cache | None = None
):
self.repository = repository
self.logger = logger
self.cache = cache
def get_user(self, user_id: str) -> User | None:
if self.cache:
cached = self.cache.get(user_id)
if cached:
return cached
user = self.repository.find_by_id(user_id)
if user and self.cache:
self.cache.set(user_id, user)
return userPackage Organization
Project Structure
project/
├── src/
│ └── mypackage/
│ ├── __init__.py
│ ├── domain/ # Business logic
│ │ ├── __init__.py
│ │ └── models.py
│ ├── services/ # Application services
│ │ ├── __init__.py
│ │ └── user_service.py
│ └── infrastructure/ # External dependencies
│ ├── __init__.py
│ └── database.py
├── tests/
│ ├── unit/
│ └── integration/
├── pyproject.toml
└── README.mdModule Exports
# __init__.py
from .models import User, Product
from .services import UserService
__all__ = ['User', 'Product', 'UserService']Error Handling
Custom Exceptions
class DomainError(Exception):
"""Base exception for domain errors"""
pass
class UserNotFoundError(DomainError):
"""Raised when user is not found"""
def __init__(self, user_id: str):
self.user_id = user_id
super().__init__(f"User {user_id} not found")
class ValidationError(DomainError):
"""Raised when validation fails"""
def __init__(self, field: str, message: str):
self.field = field
self.message = message
super().__init__(f"{field}: {message}")Exception Groups (Python 3.11+)
try:
# Multiple operations
pass
except* ValueError as eg:
# Handle all ValueError instances
for exc in eg.exceptions:
print(f"ValueError: {exc}")
except* TypeError as eg:
# Handle all TypeError instances
for exc in eg.exceptions:
print(f"TypeError: {exc}")Property Decorators
class User:
def __init__(self, name: str):
self._name = name
self._email = None
@property
def name(self) -> str:More from affaan-m
Use this skill in a project
Open any RiverX project, search the Skills panel for affaan-m/ecc/python-patterns, and install it in one click.
