
Python Type Hints
- 55 installs
- 50 repo stars
- Updated June 18, 2026
- josiahsiegel/claude-plugin-marketplace
Helps with python tasks.
About
python-type-hints is a Claude Code skill for python. It helps solo builders move faster with AI-assisted development.
- python-type-hints
- Python
- AI-coding skill
Python Type Hints by the numbers
- 55 all-time installs (skills.sh)
- +5 installs in the week ending Aug 2, 2026 (Skillselion tracking)
- Ranked #142 of 290 Python skills by installs in the Skillselion catalog
- Data as of Aug 3, 2026 (Skillselion catalog sync)
npx skills add https://github.com/josiahsiegel/claude-plugin-marketplace --skill python-type-hintsAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 55 |
|---|---|
| repo stars | ★ 50 |
| Last updated | June 18, 2026 |
| Repository | josiahsiegel/claude-plugin-marketplace ↗ |
What it does
Helps with python tasks.
Files
Quick Reference
| Type | Syntax (3.9+) | Example |
|---|---|---|
| List | list[str] | names: list[str] = [] |
| Dict | dict[str, int] | ages: dict[str, int] = {} |
| Optional | `str \ | None` |
| Union | `int \ | str` |
| Callable | Callable[[int], str] | func: Callable[[int], str] |
| Feature | Version | Syntax |
|---|---|---|
| Type params | 3.12+ | def first[T](items: list[T]) -> T: |
| type alias | 3.12+ | type Point = tuple[float, float] |
| Self | 3.11+ | def copy(self) -> Self: |
| TypeIs | 3.13+ | def is_str(x) -> TypeIs[str]: |
| Construct | Use Case |
|---|---|
Protocol | Structural subtyping (duck typing) |
TypedDict | Dict with specific keys |
Literal["a", "b"] | Specific values only |
Final[str] | Cannot be reassigned |
When to Use This Skill
Use for static type checking:
- Adding type hints to functions and classes
- Creating typed dictionaries with TypedDict
- Defining protocols for duck typing
- Configuring mypy or pyright
- Writing generic functions and classes
Related skills:
- For Python fundamentals: see
python-fundamentals-313 - For testing: see
python-testing - For FastAPI schemas: see
python-fastapi
---
Python Type Hints Complete Guide
Overview
Type hints enable static type checking, better IDE support, and self-documenting code. Python's typing system is gradual - you can add types incrementally.
Modern Type Hints (Python 3.9+)
Built-in Generic Types
# Python 3.9+ - Use built-in types directly
# No need for typing.List, typing.Dict, etc.
def process_items(items: list[str]) -> dict[str, int]:
return {item: len(item) for item in items}
# Collections
names: list[str] = ["Alice", "Bob"]
ages: dict[str, int] = {"Alice": 30, "Bob": 25}
coordinates: tuple[float, float] = (1.0, 2.0)
unique_ids: set[int] = {1, 2, 3}
frozen_data: frozenset[str] = frozenset(["a", "b"])
# Nested generics
matrix: list[list[int]] = [[1, 2], [3, 4]]
config: dict[str, list[str]] = {"servers": ["a", "b"]}Union Types (Python 3.10+)
# Old way (still works)
from typing import Union, Optional
def old_style(value: Union[int, str]) -> Optional[str]:
return str(value) if value else None
# New way (Python 3.10+)
def new_style(value: int | str) -> str | None:
return str(value) if value else None
# Optional is just Union with None
# Optional[str] == str | NoneType Aliases
# Simple type alias
UserId = int
Username = str
def get_user(user_id: UserId) -> Username:
return "user_" + str(user_id)
# Complex type alias
from typing import TypeAlias
JsonValue: TypeAlias = str | int | float | bool | None | list["JsonValue"] | dict[str, "JsonValue"]
# Python 3.12+ type statement
type Point = tuple[float, float]
type Vector[T] = list[T]
type JsonDict = dict[str, "JsonValue"]Type Parameters (Python 3.12+)
# Old way with TypeVar
from typing import TypeVar
T = TypeVar("T")
def first_old(items: list[T]) -> T:
return items[0]
# New way (Python 3.12+)
def first[T](items: list[T]) -> T:
return items[0]
# Generic classes
class Stack[T]:
def __init__(self) -> None:
self._items: list[T] = []
def push(self, item: T) -> None:
self._items.append(item)
def pop(self) -> T:
return self._items.pop()
# Multiple type parameters
def merge[K, V](d1: dict[K, V], d2: dict[K, V]) -> dict[K, V]:
return {**d1, **d2}
# Bounded type parameters
from typing import SupportsLessThan
def minimum[T: SupportsLessThan](a: T, b: T) -> T:
return a if a < b else b
# Default type parameters (Python 3.13+)
class Container[T = int]:
def __init__(self, value: T) -> None:
self.value = valueFunction Signatures
Basic Functions
from typing import Callable, Iterable, Iterator
# Simple function
def greet(name: str) -> str:
return f"Hello, {name}!"
# Multiple parameters
def create_user(name: str, age: int, email: str | None = None) -> dict:
return {"name": name, "age": age, "email": email}
# *args and **kwargs
def log(*args: str, **kwargs: int) -> None:
for arg in args:
print(arg)
for key, value in kwargs.items():
print(f"{key}={value}")
# Callable type
def apply_func(func: Callable[[int, int], int], a: int, b: int) -> int:
return func(a, b)
# Higher-order functions
def make_multiplier(n: int) -> Callable[[int], int]:
def multiplier(x: int) -> int:
return x * n
return multiplierOverloads
from typing import overload, Literal
@overload
def process(data: str) -> str: ...
@overload
def process(data: bytes) -> bytes: ...
@overload
def process(data: int) -> int: ...
def process(data: str | bytes | int) -> str | bytes | int:
if isinstance(data, str):
return data.upper()
elif isinstance(data, bytes):
return data.upper()
else:
return data * 2
# Overload with Literal
@overload
def fetch(url: str, format: Literal["json"]) -> dict: ...
@overload
def fetch(url: str, format: Literal["text"]) -> str: ...
@overload
def fetch(url: str, format: Literal["bytes"]) -> bytes: ...
def fetch(url: str, format: str) -> dict | str | bytes:
# Implementation
...ParamSpec for Decorators
from typing import ParamSpec, TypeVar, Callable
from functools import wraps
P = ParamSpec("P")
R = TypeVar("R")
def log_calls(func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
print(f"Calling {func.__name__}")
return func(*args, **kwargs)
return wrapper
@log_calls
def add(a: int, b: int) -> int:
return a + b
# Python 3.12+ syntax
def log_calls_new[**P, R](func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
print(f"Calling {func.__name__}")
return func(*args, **kwargs)
return wrapperClasses and Protocols
Class Typing
from typing import ClassVar, Self
class User:
# Class variable
count: ClassVar[int] = 0
def __init__(self, name: str, age: int) -> None:
self.name = name
self.age = age
User.count += 1
# Self type for method chaining
def with_name(self, name: str) -> Self:
self.name = name
return self
def with_age(self, age: int) -> Self:
self.age = age
return self
# Usage
user = User("Alice", 30).with_name("Bob").with_age(25)Protocols (Structural Subtyping)
from typing import Protocol, runtime_checkable
# Define a protocol (interface)
class Drawable(Protocol):
def draw(self) -> None: ...
class Resizable(Protocol):
def resize(self, width: int, height: int) -> None: ...
# Combining protocols
class DrawableAndResizable(Drawable, Resizable, Protocol):
pass
# Implementation (no explicit inheritance needed!)
class Circle:
def draw(self) -> None:
print("Drawing circle")
class Rectangle:
def draw(self) -> None:
print("Drawing rectangle")
def resize(self, width: int, height: int) -> None:
print(f"Resizing to {width}x{height}")
# Works because Circle has draw() method
def render(shape: Drawable) -> None:
shape.draw()
render(Circle()) # OK - Circle satisfies Drawable protocol
# Runtime checkable protocol
@runtime_checkable
class Closeable(Protocol):
def close(self) -> None: ...
# Can use isinstance
if isinstance(file, Closeable):
file.close()TypedDict
from typing import TypedDict, Required, NotRequired
# Basic TypedDict
class Movie(TypedDict):
title: str
year: int
director: str
movie: Movie = {"title": "Inception", "year": 2010, "director": "Nolan"}
# With optional keys
class UserProfile(TypedDict, total=False):
name: str # Optional
email: str # Optional
age: int # Optional
# Mixed required and optional (Python 3.11+)
class Article(TypedDict):
title: Required[str]
content: Required[str]
author: NotRequired[str]
tags: NotRequired[list[str]]
# Inheritance
class DetailedMovie(Movie):
rating: float
genres: list[str]Abstract Base Classes
from abc import ABC, abstractmethod
class Repository[T](ABC):
@abstractmethod
def get(self, id: int) -> T | None:
...
@abstractmethod
def save(self, entity: T) -> T:
...
@abstractmethod
def delete(self, id: int) -> bool:
...
class UserRepository(Repository["User"]):
def get(self, id: int) -> "User | None":
return self._db.get(id)
def save(self, entity: "User") -> "User":
return self._db.save(entity)
def delete(self, id: int) -> bool:
return self._db.delete(id)Advanced Types
Literal Types
from typing import Literal
# Restrict to specific values
Mode = Literal["r", "w", "a", "rb", "wb"]
def open_file(path: str, mode: Mode) -> None:
...
open_file("test.txt", "r") # OK
open_file("test.txt", "x") # Type error!
# Combining literals
HttpMethod = Literal["GET", "POST", "PUT", "DELETE", "PATCH"]
StatusCode = Literal[200, 201, 400, 401, 403, 404, 500]Final and Const
from typing import Final
# Constant that shouldn't be reassigned
MAX_SIZE: Final = 100
API_URL: Final[str] = "https://api.example.com"
# Final class methods
class Base:
from typing import final
@final
def critical_method(self) -> None:
"""Cannot be overridden in subclasses."""
...
# Final classes
from typing import final
@final
class Singleton:
"""Cannot be subclassed."""
_instance: "Singleton | None" = NoneType Guards
from typing import TypeGuard, TypeIs
# TypeGuard (narrows type in if block)
def is_string_list(val: list[object]) -> TypeGuard[list[str]]:
return all(isinstance(x, str) for x in val)
def process(items: list[object]) -> None:
if is_string_list(items):
# items is now list[str]
for item in items:
print(item.upper())
# TypeIs (Python 3.13+ - stricter than TypeGuard)
def is_int(val: int | str) -> TypeIs[int]:
return isinstance(val, int)
def handle(value: int | str) -> None:
if is_int(value):
# value is int
print(value + 1)
else:
# value is str (properly narrowed)
print(value.upper())Annotated
from typing import Annotated
from dataclasses import dataclass
# Metadata for validation/documentation
UserId = Annotated[int, "Unique user identifier"]
Email = Annotated[str, "Valid email address"]
Age = Annotated[int, "Must be >= 0"]
@dataclass
class User:
id: UserId
email: Email
age: Age
# With Pydantic
from pydantic import BaseModel, Field
class UserModel(BaseModel):
id: Annotated[int, Field(gt=0)]
email: Annotated[str, Field(pattern=r"^[\w.-]+@[\w.-]+\.\w+$")]
age: Annotated[int, Field(ge=0, le=150)]Type Checking Tools
Mypy Configuration
# pyproject.toml
[tool.mypy]
python_version = "3.12"
strict = true
warn_return_any = true
warn_unused_ignores = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
disallow_untyped_decorators = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_configs = true
# Per-module overrides
[[tool.mypy.overrides]]
module = "tests.*"
disallow_untyped_defs = false
[[tool.mypy.overrides]]
module = "third_party.*"
ignore_missing_imports = truePyright Configuration
// pyrightconfig.json
{
"include": ["src"],
"exclude": ["**/node_modules", "**/__pycache__"],
"typeCheckingMode": "strict",
"pythonVersion": "3.12",
"reportMissingImports": true,
"reportMissingTypeStubs": false,
"reportUnusedImport": true,
"reportUnusedVariable": true
}Running Type Checkers
# Mypy
mypy src/ --strict
mypy src/ --ignore-missing-imports
# Pyright (faster, VS Code default)
pyright src/
# With uv
uv run mypy src/Best Practices
1. Use Native Generics (3.9+)
# Preferred (Python 3.9+)
items: list[str] = []
mapping: dict[str, int] = {}
# Avoid (old style)
from typing import List, Dict
items: List[str] = [] # Deprecated2. Prefer Protocols Over ABCs
# Preferred - structural typing
from typing import Protocol
class Serializable(Protocol):
def to_json(self) -> str: ...
# Less flexible - nominal typing
from abc import ABC, abstractmethod
class SerializableABC(ABC):
@abstractmethod
def to_json(self) -> str: ...3. Use Abstract Collection Types
from collections.abc import Iterable, Sequence, Mapping, MutableMapping
# Prefer abstract types for function parameters
def process_items(items: Iterable[str]) -> list[str]:
return [item.upper() for item in items]
def lookup(data: Mapping[str, int], key: str) -> int | None:
return data.get(key)
# Works with any iterable/mapping
process_items(["a", "b"]) # list
process_items({"a", "b"}) # set
process_items(("a", "b")) # tuple
process_items(x for x in "ab") # generator4. Gradual Typing Strategy
# Start with public API
def public_function(data: dict[str, Any]) -> list[str]:
return _internal_helper(data)
# Type internal helpers later
def _internal_helper(data): # Untyped initially
...
# Aim for 80%+ coverage on new code
# Use # type: ignore sparingly5. Document Complex Types
from typing import TypeAlias
# Use type aliases for complex types
JsonPrimitive: TypeAlias = str | int | float | bool | None
JsonArray: TypeAlias = list["JsonValue"]
JsonObject: TypeAlias = dict[str, "JsonValue"]
JsonValue: TypeAlias = JsonPrimitive | JsonArray | JsonObject
def parse_json(text: str) -> JsonValue:
"""Parse JSON string into typed Python value."""
import json
return json.loads(text)Additional References
For advanced typing patterns beyond this guide, see:
- [Advanced Typing Patterns](references/advanced-typing-patterns.md) - Generic repository pattern, discriminated unions, builder pattern with Self, ParamSpec decorators, conditional types with overloads, typed decorator factories, Protocols with class methods, typed context variables, recursive types, typed event systems
Advanced Typing Patterns
Production-ready typing patterns for complex Python applications.
Generic Repository Pattern
from typing import TypeVar, Generic, Protocol
from abc import abstractmethod
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, delete, update
from sqlalchemy.orm import DeclarativeBase
T = TypeVar("T", bound=DeclarativeBase)
class Repository(Protocol[T]):
"""Repository protocol for data access."""
async def get(self, id: int) -> T | None: ...
async def get_all(self) -> list[T]: ...
async def create(self, entity: T) -> T: ...
async def update(self, id: int, **fields) -> T | None: ...
async def delete(self, id: int) -> bool: ...
class SQLAlchemyRepository(Generic[T]):
"""SQLAlchemy implementation of repository."""
def __init__(self, session: AsyncSession, model: type[T]):
self.session = session
self.model = model
async def get(self, id: int) -> T | None:
return await self.session.get(self.model, id)
async def get_all(self) -> list[T]:
result = await self.session.execute(select(self.model))
return list(result.scalars().all())
async def create(self, entity: T) -> T:
self.session.add(entity)
await self.session.commit()
await self.session.refresh(entity)
return entity
async def update(self, id: int, **fields) -> T | None:
await self.session.execute(
update(self.model).where(self.model.id == id).values(**fields)
)
await self.session.commit()
return await self.get(id)
async def delete(self, id: int) -> bool:
result = await self.session.execute(
delete(self.model).where(self.model.id == id)
)
await self.session.commit()
return result.rowcount > 0
# Type-safe usage
class UserRepository(SQLAlchemyRepository[User]):
"""User-specific repository with additional methods."""
async def get_by_email(self, email: str) -> User | None:
result = await self.session.execute(
select(self.model).where(self.model.email == email)
)
return result.scalar_one_or_none()
# Usage
repo: Repository[User] = UserRepository(session, User)
user = await repo.get(1) # Returns User | NoneDiscriminated Unions
from typing import Literal, Union
from dataclasses import dataclass
@dataclass
class EmailNotification:
type: Literal["email"] = "email"
to: str
subject: str
body: str
@dataclass
class SMSNotification:
type: Literal["sms"] = "sms"
to: str
message: str
@dataclass
class PushNotification:
type: Literal["push"] = "push"
device_token: str
title: str
body: str
Notification = Union[EmailNotification, SMSNotification, PushNotification]
async def send_notification(notification: Notification) -> bool:
match notification.type:
case "email":
return await send_email(notification.to, notification.subject, notification.body)
case "sms":
return await send_sms(notification.to, notification.message)
case "push":
return await send_push(notification.device_token, notification.title, notification.body)
# Type-safe - mypy knows the specific type in each branch
def format_notification(notification: Notification) -> str:
match notification:
case EmailNotification(to=to, subject=subject):
return f"Email to {to}: {subject}"
case SMSNotification(to=to, message=message):
return f"SMS to {to}: {message}"
case PushNotification(device_token=token, title=title):
return f"Push to {token}: {title}"Builder Pattern with Type Safety
from typing import TypeVar, Generic, Self
from dataclasses import dataclass, field
@dataclass
class QueryBuilder[T]:
"""Type-safe query builder."""
_model: type[T]
_filters: list[str] = field(default_factory=list)
_order_by: str | None = None
_limit: int | None = None
def filter(self, condition: str) -> Self:
self._filters.append(condition)
return self
def order_by(self, field: str) -> Self:
self._order_by = field
return self
def limit(self, n: int) -> Self:
self._limit = n
return self
def build(self) -> str:
query = f"SELECT * FROM {self._model.__tablename__}"
if self._filters:
query += " WHERE " + " AND ".join(self._filters)
if self._order_by:
query += f" ORDER BY {self._order_by}"
if self._limit:
query += f" LIMIT {self._limit}"
return query
# Usage
query = (
QueryBuilder(User)
.filter("is_active = true")
.filter("role = 'admin'")
.order_by("created_at DESC")
.limit(10)
.build()
)Callback Types with ParamSpec
from typing import ParamSpec, TypeVar, Callable, Awaitable
from functools import wraps
import time
P = ParamSpec("P")
R = TypeVar("R")
def timing(func: Callable[P, R]) -> Callable[P, R]:
"""Preserve function signature in decorator."""
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} took {elapsed:.4f}s")
return result
return wrapper
def async_timing(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
"""Async version preserving signature."""
@wraps(func)
async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
start = time.perf_counter()
result = await func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} took {elapsed:.4f}s")
return result
return wrapper
# Usage - type checker knows exact signature
@timing
def calculate(x: int, y: int, *, precision: int = 2) -> float:
return round(x / y, precision)
result = calculate(10, 3, precision=4) # Type: floatConditional Types with Overloads
from typing import overload, Literal
@overload
def fetch(url: str, *, as_json: Literal[True]) -> dict: ...
@overload
def fetch(url: str, *, as_json: Literal[False] = False) -> str: ...
def fetch(url: str, *, as_json: bool = False) -> dict | str:
response = requests.get(url)
if as_json:
return response.json()
return response.text
# Type checker knows the exact return type
data: dict = fetch("https://api.example.com", as_json=True)
text: str = fetch("https://api.example.com")
@overload
def get_item(items: list[int], index: int) -> int: ...
@overload
def get_item(items: list[str], index: int) -> str: ...
@overload
def get_item[T](items: list[T], index: int) -> T: ...
def get_item(items: list, index: int):
return items[index]
# Type preserved
x: int = get_item([1, 2, 3], 0)
s: str = get_item(["a", "b"], 0)Typed Decorators Factory
from typing import TypeVar, Callable, ParamSpec, overload
from functools import wraps
P = ParamSpec("P")
R = TypeVar("R")
def retry(
max_attempts: int = 3,
exceptions: tuple[type[Exception], ...] = (Exception,)
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""Retry decorator with preserved types."""
def decorator(func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
last_exception: Exception | None = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
raise last_exception or RuntimeError("Retry failed")
return wrapper
return decorator
@retry(max_attempts=3, exceptions=(ValueError, IOError))
def fetch_data(url: str, timeout: int = 30) -> dict:
# Implementation
...
# Type checker knows: fetch_data(url: str, timeout: int = 30) -> dictProtocol with Class Methods
from typing import Protocol, Self, ClassVar
class Serializable(Protocol):
"""Protocol for serializable objects."""
@classmethod
def from_dict(cls, data: dict) -> Self: ...
def to_dict(self) -> dict: ...
class JSONSerializable(Protocol):
"""Protocol with class variables."""
json_fields: ClassVar[list[str]]
def to_json(self) -> str: ...
@dataclass
class User:
"""User implementing both protocols."""
json_fields: ClassVar[list[str]] = ["id", "name", "email"]
id: int
name: str
email: str
@classmethod
def from_dict(cls, data: dict) -> "User":
return cls(**data)
def to_dict(self) -> dict:
return {"id": self.id, "name": self.name, "email": self.email}
def to_json(self) -> str:
return json.dumps(self.to_dict())
def serialize(obj: Serializable) -> dict:
return obj.to_dict()
def deserialize[T: Serializable](cls: type[T], data: dict) -> T:
return cls.from_dict(data)
# Usage
user_dict = serialize(User(1, "John", "john@example.com"))
user = deserialize(User, user_dict)Typed Context Variables
from contextvars import ContextVar
from typing import Generic, TypeVar
T = TypeVar("T")
class TypedContextVar(Generic[T]):
"""Type-safe context variable wrapper."""
def __init__(self, name: str, default: T | None = None):
self._var: ContextVar[T | None] = ContextVar(name, default=default)
def get(self) -> T:
value = self._var.get()
if value is None:
raise RuntimeError(f"Context variable not set")
return value
def get_or_default(self, default: T) -> T:
return self._var.get() or default
def set(self, value: T) -> None:
self._var.set(value)
# Usage
request_id: TypedContextVar[str] = TypedContextVar("request_id")
current_user: TypedContextVar[User] = TypedContextVar("current_user")
# Type-safe
request_id.set("abc-123")
user = current_user.get() # Returns User, raises if not setRecursive Types
from typing import TypeAlias
# JSON type (recursive)
JsonPrimitive: TypeAlias = str | int | float | bool | None
JsonArray: TypeAlias = list["JsonValue"]
JsonObject: TypeAlias = dict[str, "JsonValue"]
JsonValue: TypeAlias = JsonPrimitive | JsonArray | JsonObject
def parse_json(text: str) -> JsonValue:
import json
return json.loads(text)
# Tree structure
@dataclass
class TreeNode[T]:
value: T
children: list["TreeNode[T]"] = field(default_factory=list)
def add_child(self, value: T) -> "TreeNode[T]":
child = TreeNode(value)
self.children.append(child)
return child
def traverse(self) -> Iterator[T]:
yield self.value
for child in self.children:
yield from child.traverse()
# Usage
root: TreeNode[str] = TreeNode("root")
child1 = root.add_child("child1")
child1.add_child("grandchild1")
for value in root.traverse():
print(value) # Type: strTyped Event System
from typing import TypeVar, Generic, Callable, Awaitable
from dataclasses import dataclass, field
E = TypeVar("E")
@dataclass
class Event:
"""Base event class."""
timestamp: float = field(default_factory=time.time)
@dataclass
class UserCreatedEvent(Event):
user_id: int
email: str
@dataclass
class OrderPlacedEvent(Event):
order_id: int
user_id: int
total: float
class TypedEventBus:
"""Type-safe event bus."""
def __init__(self):
self._handlers: dict[type, list[Callable]] = {}
def subscribe[E: Event](
self,
event_type: type[E],
handler: Callable[[E], Awaitable[None]]
) -> None:
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
async def publish[E: Event](self, event: E) -> None:
handlers = self._handlers.get(type(event), [])
await asyncio.gather(*[h(event) for h in handlers])
# Usage
bus = TypedEventBus()
async def on_user_created(event: UserCreatedEvent) -> None:
print(f"User {event.user_id} created with email {event.email}")
bus.subscribe(UserCreatedEvent, on_user_created)
await bus.publish(UserCreatedEvent(user_id=1, email="test@example.com"))Related skills
Pythonbackend