Transport-agnostic, async-first Python service framework built around:
- typed command/query/event contracts
- modular composition
- execution middleware + lifecycle hooks
- pluggable transports (HTTP, jobs/worker, RPC, scheduler)
Phase 1-3 implementation is complete according to repository audits:
Full user docs are in docs/.
Recommended start:
- Quickstart
- Core Concepts
- HTTP, Auth, OpenAPI
- Persistence (SQLAlchemy)
- Background and Distributed Runtime
- Testing
- Configuration
- CLI Reference
- Operations and Tuning
pip install -e ".[dev,http]"from pyferox.core import App, Module, StructCommand, StructQuery, handle, singleton
from pyferox.http import HTTPAdapter
class UserRepo:
async def create(self, email: str, name: str) -> int:
return 1
async def get(self, user_id: int) -> dict[str, str]:
return {"id": str(user_id), "email": "user@example.com", "name": "User"}
class CreateUser(StructCommand):
email: str
name: str
class GetUser(StructQuery):
user_id: int
@handle(CreateUser)
async def create_user(cmd: CreateUser, users: UserRepo) -> dict[str, int]:
return {"id": await users.create(cmd.email, cmd.name)}
@handle(GetUser)
async def get_user(query: GetUser, users: UserRepo) -> dict[str, str]:
return await users.get(query.user_id)
app = App(modules=[Module(handlers=[create_user, get_user], providers=[singleton(UserRepo())])])
http = HTTPAdapter(app)
http.command("POST", "/users", CreateUser, status_code=201)
http.query("GET", "/users/{user_id}", GetUser)PyFerOx is msgspec-backed. For new transport-facing contracts, prefer StructCommand / StructQuery / StructEvent. Dataclass contracts are still supported, but they are not the primary style to teach in user-facing examples.
Run:
uvicorn app.main:http --reloadOr scaffold first:
pyferox create-project demo_service --template api
cd demo_service
pyferox run-dev --target app.main:http