fastapi-pro

Name: fastapi-pro
Author: tranhieutt

69stars41forks

How to install fastapi-pro?

tranhieutt/software_development_department/fastapi-pro

$npx -y skills add tranhieutt/software_development_department --skill fastapi-pro

Installs into the current project.

Summary

Production FastAPI patterns — async endpoints, SQLAlchemy 2.0 async, Pydantic V2, dependency injection, JWT auth, testing. Use for Python 3.11+ FastAPI backends. NOT for Django (→ django-patterns) or Node.js (→ backend-patterns).

Files · 1

View on GitHub

SKILL.md

1	# FastAPI Production Patterns
2
3	## Critical rules (non-obvious)
4
5	- `async def` endpoint blocking sync DB call → blocks entire event loop. Either use `async` DB driver (asyncpg/aiomysql) throughout OR switch endpoint to plain `def` (FastAPI runs it in threadpool).
6	- Pydantic V2 `model_config = ConfigDict(...)` replaces V1 `class Config`. Forgetting this silently loses settings like `from_attributes=True` needed for ORM → DTO conversion.
7	- `Depends()` caches per-request: same dependency called twice in one request returns same instance. Don't rely on this for cross-request state — use app state / Redis instead.
8	- SQLAlchemy 2.0 async session must not leak across requests: always scope via `Depends` with `async with AsyncSession(...)` — raw module-level session causes `GreenletError` under load.
9	- `BackgroundTasks` runs AFTER response sent in the same worker process: if worker dies mid-task the work is lost. For durable background jobs use Celery / Dramatiq / ARQ.
10	- Uvicorn `--workers N` forks processes — can't share in-memory state. Use Redis or DB for any shared state (rate-limit counters, cache).
11
12	## Project layout
13
14	```
15	app/
16	├── main.py # FastAPI() instance + lifespan
17	├── api/
18	│ ├── deps.py # shared Depends (get_db, get_current_user)
19	│ └── v1/
20	│ ├── users.py # APIRouter
21	│ └── products.py
22	├── core/
23	│ ├── config.py # Pydantic Settings
24	│ ├── security.py # JWT encode/decode, password hashing
25	│ └── db.py # engine + AsyncSession factory
26	├── models/ # SQLAlchemy ORM models
27	├── schemas/ # Pydantic DTOs (Request/Response)
28	├── services/ # business logic (no framework coupling)
29	└── tests/
30	```
31
32	## Pydantic V2 settings + config
33
34	```python
35	# app/core/config.py
36	from pydantic import Field
37	from pydantic_settings import BaseSettings, SettingsConfigDict
38
39	class Settings(BaseSettings):
40	model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_")
41
42	database_url: str = Field(..., description="postgresql+asyncpg://...")
43	jwt_secret: str = Field(..., min_length=32)
44	jwt_algorithm: str = "HS256"
45	jwt_exp_minutes: int = 30
46	cors_origins: list[str] = Field(default_factory=list)
47
48	settings = Settings() # fails fast at import if required vars missing
49	```
50
51	## SQLAlchemy 2.0 async session (per-request)
52
53	```python
54	# app/core/db.py
55	from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
56
57	engine = create_async_engine(
58	settings.database_url,
59	pool_size=20,
60	max_overflow=10,
61	pool_pre_ping=True, # reconnect on stale conns (LB idle timeout)
62	echo=False,
63	)
64	SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
65
66	# app/api/deps.py
67	from typing import AsyncIterator
68	from fastapi import Depends
69
70	async def get_db() -> AsyncIterator[AsyncSession]:
71	async with SessionLocal() as session:
72	try:
73	yield session
74	await session.commit()
75	except Exception:
76	await session.rollback()
77	raise
78	# close is automatic via `async with`
79	```
80
81	## Lifespan + startup/shutdown
82
83	```python
84	# app/main.py
85	from contextlib import asynccontextmanager
86	from fastapi import FastAPI
87
88	@asynccontextmanager
89	async def lifespan(app: FastAPI):
90	# Startup: warm caches, test DB
91	async with engine.begin() as conn:
92	await conn.execute(text("SELECT 1"))
93	yield
94	# Shutdown: drain connections
95	await engine.dispose()
96
97	app = FastAPI(title="My API", lifespan=lifespan)
98	```
99
100	## JWT auth with OAuth2PasswordBearer
101
102	```python
103	# app/core/security.py
104	from datetime import datetime, timedelta, timezone
105	from jose import jwt, JWTError
106	from passlib.context import CryptContext
107
108	pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
109
110	def hash_password(p: str) -> str: return pwd.hash(p)
111	def verify_password(p: str, h: str) -> bool: return pwd.verify(p, h)
112
113	def create_access_token(sub: str) -> str:
114	exp = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_exp_minutes)
115	return jwt.encode({"sub": sub, "exp": exp}, settings.jwt_secret, settings.jwt_algorithm)
116
117	# app/api/deps.py
118	from fastapi import HTTPException, status
119	from fastapi.security import OAuth2PasswordBearer
120
121	oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
122
123	async def get_current_

People who liked this also viewed

6 picks

Type

fastapi-pro

69stars41forks

View on GitHub

How to install fastapi-pro?

tranhieutt/software_development_department/fastapi-pro

$npx -y skills add tranhieutt/software_development_department --skill fastapi-pro

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# FastAPI Production Patterns
2
3	## Critical rules (non-obvious)
4
5	- `async def` endpoint blocking sync DB call → blocks entire event loop. Either use `async` DB driver (asyncpg/aiomysql) throughout OR switch endpoint to plain `def` (FastAPI runs it in threadpool).
6	- Pydantic V2 `model_config = ConfigDict(...)` replaces V1 `class Config`. Forgetting this silently loses settings like `from_attributes=True` needed for ORM → DTO conversion.
7	- `Depends()` caches per-request: same dependency called twice in one request returns same instance. Don't rely on this for cross-request state — use app state / Redis instead.
8	- SQLAlchemy 2.0 async session must not leak across requests: always scope via `Depends` with `async with AsyncSession(...)` — raw module-level session causes `GreenletError` under load.
9	- `BackgroundTasks` runs AFTER response sent in the same worker process: if worker dies mid-task the work is lost. For durable background jobs use Celery / Dramatiq / ARQ.
10	- Uvicorn `--workers N` forks processes — can't share in-memory state. Use Redis or DB for any shared state (rate-limit counters, cache).
11
12	## Project layout
13
14	```
15	app/
16	├── main.py # FastAPI() instance + lifespan
17	├── api/
18	│ ├── deps.py # shared Depends (get_db, get_current_user)
19	│ └── v1/
20	│ ├── users.py # APIRouter
21	│ └── products.py
22	├── core/
23	│ ├── config.py # Pydantic Settings
24	│ ├── security.py # JWT encode/decode, password hashing
25	│ └── db.py # engine + AsyncSession factory
26	├── models/ # SQLAlchemy ORM models
27	├── schemas/ # Pydantic DTOs (Request/Response)
28	├── services/ # business logic (no framework coupling)
29	└── tests/
30	```
31
32	## Pydantic V2 settings + config
33
34	```python
35	# app/core/config.py
36	from pydantic import Field
37	from pydantic_settings import BaseSettings, SettingsConfigDict
38
39	class Settings(BaseSettings):
40	model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_")
41
42	database_url: str = Field(..., description="postgresql+asyncpg://...")
43	jwt_secret: str = Field(..., min_length=32)
44	jwt_algorithm: str = "HS256"
45	jwt_exp_minutes: int = 30
46	cors_origins: list[str] = Field(default_factory=list)
47
48	settings = Settings() # fails fast at import if required vars missing
49	```
50
51	## SQLAlchemy 2.0 async session (per-request)
52
53	```python
54	# app/core/db.py
55	from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
56
57	engine = create_async_engine(
58	settings.database_url,
59	pool_size=20,
60	max_overflow=10,
61	pool_pre_ping=True, # reconnect on stale conns (LB idle timeout)
62	echo=False,
63	)
64	SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
65
66	# app/api/deps.py
67	from typing import AsyncIterator
68	from fastapi import Depends
69
70	async def get_db() -> AsyncIterator[AsyncSession]:
71	async with SessionLocal() as session:
72	try:
73	yield session
74	await session.commit()
75	except Exception:
76	await session.rollback()
77	raise
78	# close is automatic via `async with`
79	```
80
81	## Lifespan + startup/shutdown
82
83	```python
84	# app/main.py
85	from contextlib import asynccontextmanager
86	from fastapi import FastAPI
87
88	@asynccontextmanager
89	async def lifespan(app: FastAPI):
90	# Startup: warm caches, test DB
91	async with engine.begin() as conn:
92	await conn.execute(text("SELECT 1"))
93	yield
94	# Shutdown: drain connections
95	await engine.dispose()
96
97	app = FastAPI(title="My API", lifespan=lifespan)
98	```
99
100	## JWT auth with OAuth2PasswordBearer
101
102	```python
103	# app/core/security.py
104	from datetime import datetime, timedelta, timezone
105	from jose import jwt, JWTError
106	from passlib.context import CryptContext
107
108	pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
109
110	def hash_password(p: str) -> str: return pwd.hash(p)
111	def verify_password(p: str, h: str) -> bool: return pwd.verify(p, h)
112
113	def create_access_token(sub: str) -> str:
114	exp = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_exp_minutes)
115	return jwt.encode({"sub": sub, "exp": exp}, settings.jwt_secret, settings.jwt_algorithm)
116
117	# app/api/deps.py
118	from fastapi import HTTPException, status
119	from fastapi.security import OAuth2PasswordBearer
120
121	oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
122
123	async def get_current_

People who liked this also viewed

6 picks

Type