patterns#

class design, decorators, error handling, and other structural patterns.

private internals#

keep implementation details in _internal/:

src/mypackage/
├── __init__.py          # public API
├── cli.py               # entry point
└── _internal/           # implementation details
    ├── config.py
    ├── operations.py
    └── types.py

__init__.py re-exports the public interface. users import from the package, not from internals.

dataclasses for DTOs#

simple data containers that don't need validation:

from dataclasses import dataclass

@dataclass
class BatchResult:
    successful: list[str]
    failed: list[tuple[str, Exception]]

    @property
    def total(self) -> int:
        return len(self.successful) + len(self.failed)

lighter than pydantic when you control the data source.

from pdsx/_internal/batch.py

base classes with parallel implementations#

when you need both sync and async:

class _BaseClient:
    def __init__(self, *, token: str | None = None):
        self._token = token or get_settings().token

class Client(_BaseClient):
    def get(self, url: str) -> dict:
        return httpx.get(url, headers=self._headers).json()

class AsyncClient(_BaseClient):
    async def get(self, url: str) -> dict:
        async with httpx.AsyncClient() as client:
            return (await client.get(url, headers=self._headers)).json()

shared logic in base, divergent implementation in subclasses.

fluent interfaces#

chainable methods that return self:

class Track:
    def __init__(self, source: str):
        self._source = source
        self._effects: list[str] = []

    def volume(self, level: float) -> "Track":
        self._effects.append(f"volume={level}")
        return self

    def lowpass(self, freq: float) -> "Track":
        self._effects.append(f"lowpass=f={freq}")
        return self

# usage
track.volume(0.8).lowpass(600).fade_in(0.5)

factory classmethods#

alternative constructors:

@dataclass
class URIParts:
    """parsed components of an AT-URI."""
    repo: str
    collection: str
    rkey: str

    @classmethod
    def from_uri(cls, uri: str, client_did: str | None = None) -> URIParts:
        """parse an AT-URI into its components."""
        uri_without_prefix = uri.replace("at://", "")
        parts = uri_without_prefix.split("/")

        # shorthand format: collection/rkey
        if len(parts) == 2:
            if not client_did:
                raise ValueError("shorthand URI requires authentication")
            return cls(repo=client_did, collection=parts[0], rkey=parts[1])

        # full format: did/collection/rkey
        if len(parts) == 3:
            return cls(repo=parts[0], collection=parts[1], rkey=parts[2])

        raise ValueError(f"invalid URI format: {uri}")

from pdsx/_internal/resolution.py

keyword-only arguments#

force callers to name arguments for clarity:

def batch_create(
    client: Client,
    collection: str,
    records: list[dict],
    *,  # everything after is keyword-only
    concurrency: int = 10,
    fail_fast: bool = False,
) -> BatchResult:
    ...

# must use: batch_create(client, "posts", items, concurrency=5)
# not: batch_create(client, "posts", items, 5)

custom exceptions#

with helpful context:

class AuthenticationRequired(Exception):
    def __init__(self, operation: str = "this operation"):
        super().__init__(
            f"{operation} requires authentication.\n\n"
            "Set ATPROTO_HANDLE and ATPROTO_PASSWORD environment variables."
        )

the message tells you what to do, not just what went wrong.

from pdsx/mcp/client.py

exception hierarchies#

for structured error handling:

class AppError(Exception):
    """base for all app errors."""

class ValidationError(AppError):
    """input validation failed."""

class ResourceError(AppError):
    """resource operation failed."""

# callers can catch AppError for all, or specific types

argparse for CLIs#

argparse is stdlib. subparsers with aliases give you unix-style commands:

parser = argparse.ArgumentParser(
    description="my tool",
    formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("-r", "--repo", help="target repo")

subparsers = parser.add_subparsers(dest="command")

list_parser = subparsers.add_parser("list", aliases=["ls"])
list_parser.add_argument("collection")
list_parser.add_argument("--limit", type=int, default=50)

dispatch with tuple membership to handle aliases:

args = parser.parse_args()

if not args.command:
    parser.print_help()
    return 1

if args.command in ("list", "ls"):
    return await cmd_list(args.collection, args.limit)

async entry point pattern:

async def async_main() -> int:
    # argparse and dispatch here
    return 0

def main() -> NoReturn:
    sys.exit(asyncio.run(async_main()))

source: pdsx/src/pdsx/cli.py

module-level singletons#

instantiate once, import everywhere:

# config.py
from functools import lru_cache

@lru_cache
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

# console.py
from rich.console import Console

console = Console()

sources: