Python Design Snippets - Additional snippets from a post on designing python programs

composition_example.py

import os
from functools import partial
from typing import Any, Callable, Dict, Tuple

# Type Aliases
LogMap = Dict[str, str]

# Constant values
DEFAULTS: Dict[str, Any] = {
    "ALERT_THRESHOLD": 2,
    "LOG_LEVELS": ("DEBUG", "INFO", "WARN", "ERROR"),
    "LOG_FORMAT": "timestamp,application,log_level,message",
    "FORMAT_DELIM": ",",
    "LOG_DELIM": "|",
}

# These are some building block functions
def _opt(name: str, defaults: Dict[str, Any]) -> Any:
    return os.environ.get(name, defaults[name])


def _split(raw_str: str, delim: str) -> Tuple[str]:
    return tuple(raw_str.split(delim))


def _join(log: LogMap, delim: str) -> str:
    return delim.join(log.values())


def _parse(log: Tuple[str], log_format: Tuple[str]) -> LogMap:
    return {field.upper(): value for field, value in zip(log_format, log)}


# Example Processor
def _alert(log: LogMap, levels: Tuple[str], threshold: int) -> LogMap:
    if levels.index(log["LOG_LEVEL"]) >= threshold:
        print("ALERTING ON THIS LOG! ALERT!")
    return log


def get_pipeline() -> Callable[[str], str]:
    """
    Composes building blocks to create a log processing pipeline.
    """
    # Compose Partials
    get_opt = partial(_opt, defaults=DEFAULTS)
    log_splitter = partial(_split, delim=get_opt("LOG_DELIM"))
    format_splitter = partial(_split, delim=get_opt("FORMAT_DELIM"))
    joiner = partial(_join, delim=get_opt("LOG_DELIM"))
    alert = partial(_alert, levels=get_opt("LOG_LEVELS"), threshold=get_opt("ALERT_THRESHOLD"))
    log_format =format_splitter(get_opt("LOG_FORMAT"))  # Only calculate format once

    # Compose with lambda because arguments need processing
    parser = lambda log: _parse(splitter(log), log_format=log_format)
    pipeline = lambda log: joiner(alert(parser(log)))

    return pipeline


def main():
    """Example usage"""
    pipeline = get_pipeline()
    example_log = "1234567|myapp|WARN|The British are coming!"
    print(f"{example_log} -> {pipeline(example_log)}")


if __name__ == "__main__":
    main()

namespace_example.py

"""This module is used to demonstrate the concept of namespaces"""
print(f"Initial Namespace: {dir()}")
# >> ['__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__spec__']

print(f"'requests' in namespace? {'requests' in dir()}") # >> False
import requests
print(f"'requests' in namespace? {'requests' in dir()}") # >> True


def connect(host: str) -> requests.Response:
    response = requests.post(host, json={"hello": "dad"})
    return response
print(f"'connect' in namespace? {'connect' in dir()}") ## >> True

x = 5
print(f"'x' in namespace? {'x' in dir()}") # >> True

print(f"Final Namespace: {dir()}")
# >> ['__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__spec__', 'connect', 'requests', 'x']

state_example.py

from typing import ClassVar, Dict

class PersonMeta(type):
    """Spooky metaclass that tracks all people."""
    _national_registry: ClassVar[Dict[str, "Person"]] = {}

    def __call__(cls: "Person", *args, **kwargs):
        """Hooks class instantiation."""
        instance = super().__call__(*args, **kwargs)
        PersonMeta._national_registry[instance.name] = instance
        instance.ssn = len(PersonMeta._national_registry)
        return instance

    @classmethod
    def lookup(cls, name: str) -> "Person":
        """Lookup a person by name and return their instance."""
        return PersonMeta._national_registry[name]

class Person(metaclass=PersonMeta):
    """Represents your average joe."""
    # Class Attribute: All Persons are human
    species: ClassVar[str] = "human"

    # Instance Attribute: Each person has their own name and ssn
    name: str
    ssn: int

    def __init__(self, name: str):
        self.name = name

if __name__ == "__main__":
    print("State Demo")
    print(f"All people are {Person.species}")
    joe = Person("Joe")
    print(f"One is named {joe.name}")
    assert joe is Person.lookup("Joe")
    print(f"And he is tracked... SSN={Person.lookup('Joe').ssn}")