scaffolding-fastapi-dapr

Name: scaffolding-fastapi-dapr
Author: mjunaidca

23installs11forks

How to install scaffolding-fastapi-dapr?

mjunaidca/mjs-agent-skills/scaffolding-fastapi-dapr

$npx -y skills add mjunaidca/mjs-agent-skills --skill scaffolding-fastapi-dapr

Installs into the current project.

Summary

Build production-grade FastAPI backends with SQLModel, Dapr integration, and JWT authentication. Use when building REST APIs with Neon PostgreSQL, implementing event-driven microservices with Dapr pub/sub, scheduling jobs, or creating CRUD endpoints with JWT/JWKS verification. NO

Files · 1

View on GitHub

SKILL.md

1	# FastAPI + Dapr Backend
2
3	Build production-grade FastAPI backends with SQLModel, Dapr integration, and JWT authentication.
4
5	## Quick Start
6
7	```bash
8	# Project setup
9	uv init backend && cd backend
10	uv add fastapi sqlmodel pydantic httpx python-jose uvicorn
11
12	# Development
13	uv run uvicorn main:app --reload --port 8000
14
15	# With Dapr sidecar
16	dapr run --app-id myapp --app-port 8000 -- uvicorn main:app
17	```
18
19	---
20
21	## FastAPI Core Patterns
22
23	### 1. SQLModel Schema (Database + API)
24
25	```python
26	from sqlmodel import SQLModel, Field
27	from datetime import datetime
28	from typing import Optional, Literal
29
30	class TaskBase(SQLModel):
31	title: str = Field(max_length=200, index=True)
32	status: Literal["pending", "in_progress", "completed"] = "pending"
33
34	class Task(TaskBase, table=True):
35	id: Optional[int] = Field(default=None, primary_key=True)
36	created_at: datetime = Field(default_factory=datetime.now)
37
38	class TaskCreate(TaskBase):
39	pass
40
41	class TaskRead(TaskBase):
42	id: int
43	created_at: datetime
44	```
45
46	### 2. Async Database Setup
47
48	```python
49	from sqlmodel.ext.asyncio.session import AsyncSession
50	from sqlalchemy.ext.asyncio import create_async_engine
51	import os
52
53	DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
54	engine = create_async_engine(DATABASE_URL)
55
56	async def get_session() -> AsyncSession:
57	async with AsyncSession(engine) as session:
58	yield session
59	```
60
61	### 3. CRUD Endpoints
62
63	```python
64	from fastapi import FastAPI, Depends, HTTPException
65	from sqlmodel import select
66
67	app = FastAPI()
68
69	@app.post("/tasks", response_model=TaskRead, status_code=201)
70	async def create_task(task: TaskCreate, session: AsyncSession = Depends(get_session)):
71	db_task = Task.model_validate(task)
72	session.add(db_task)
73	await session.commit()
74	await session.refresh(db_task)
75	return db_task
76
77	@app.get("/tasks/{task_id}", response_model=TaskRead)
78	async def get_task(task_id: int, session: AsyncSession = Depends(get_session)):
79	task = await session.get(Task, task_id)
80	if not task:
81	raise HTTPException(status_code=404, detail="Not found")
82	return task
83
84	@app.patch("/tasks/{task_id}", response_model=TaskRead)
85	async def update_task(task_id: int, update: TaskUpdate, session: AsyncSession = Depends(get_session)):
86	task = await session.get(Task, task_id)
87	if not task:
88	raise HTTPException(status_code=404, detail="Not found")
89	update_data = update.model_dump(exclude_unset=True)
90	task.sqlmodel_update(update_data)
91	session.add(task)
92	await session.commit()
93	await session.refresh(task)
94	return task
95	```
96
97	### 4. JWT/JWKS Authentication
98
99	```python
100	from jose import jwt
101	import httpx
102
103	JWKS_URL = f"{SSO_URL}/.well-known/jwks.json"
104
105	async def get_current_user(authorization: str = Header()):
106	token = authorization.replace("Bearer ", "")
107	async with httpx.AsyncClient() as client:
108	jwks = (await client.get(JWKS_URL)).json()
109	payload = jwt.decode(token, jwks, algorithms=["RS256"])
110	return payload
111
112	@app.get("/protected")
113	async def protected_route(user = Depends(get_current_user)):
114	return {"user": user["sub"]}
115	```
116
117	See [references/fastapi-patterns.md](references/fastapi-patterns.md) for audit logging, pagination, and OpenAPI configuration.
118
119	---
120
121	## Dapr Integration Patterns
122
123	### 1. Pub/Sub Subscription
124
125	```python
126	from fastapi import APIRouter, Request
127
128	router = APIRouter(prefix="/dapr", tags=["Dapr"])
129
130	@router.get("/subscribe")
131	async def subscribe():
132	"""Dapr calls this to discover subscriptions."""
133	return [{
134	"pubsubname": "pubsub",
135	"topic": "task-created",
136	"route": "/dapr/task-created"
137	}]
138
139	@router.post("/task-created")
140	async def handle_task_created(request: Request, session: AsyncSession = Depends(get_session)):
141	# CloudEvent wrapper - data is nested
142	event = await request.json()
143	task_data = event.get("data", event) # Handle both wrapped and unwrapped
144
145	# Process event
146	task = Task.model_validate(task_data)
147	session.add(task)
148	await session.commit()
149	return {"status": "processed"}
150	```
151
152	### 2. Publishing Events
153
154	```python
155	import httpx
156
157	DAPR_URL = "http://localhost:3500"
158
159	async def publish_event(topic: str, data: dict):
160	async with httpx.AsyncClient() as client:
161	await client.post(
162	f"{DAPR_URL}/v1.0/publish/pubsub/{topic}",
163	json=data,
164	headers={"Content-Type": "application/json"}
165	)
166	```
167
168	### 3. Scheduled Jobs
169
170	```python
171	# Schedule a job via Dapr Jobs API (alpha)
172	async def schedule_job(name: str, schedule: str, callback_url: str, data: dict):

People who liked this also viewed

6 picks

Type

scaffolding-fastapi-dapr

23installs11forks

View on GitHub

How to install scaffolding-fastapi-dapr?

mjunaidca/mjs-agent-skills/scaffolding-fastapi-dapr

$npx -y skills add mjunaidca/mjs-agent-skills --skill scaffolding-fastapi-dapr

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# FastAPI + Dapr Backend
2
3	Build production-grade FastAPI backends with SQLModel, Dapr integration, and JWT authentication.
4
5	## Quick Start
6
7	```bash
8	# Project setup
9	uv init backend && cd backend
10	uv add fastapi sqlmodel pydantic httpx python-jose uvicorn
11
12	# Development
13	uv run uvicorn main:app --reload --port 8000
14
15	# With Dapr sidecar
16	dapr run --app-id myapp --app-port 8000 -- uvicorn main:app
17	```
18
19	---
20
21	## FastAPI Core Patterns
22
23	### 1. SQLModel Schema (Database + API)
24
25	```python
26	from sqlmodel import SQLModel, Field
27	from datetime import datetime
28	from typing import Optional, Literal
29
30	class TaskBase(SQLModel):
31	title: str = Field(max_length=200, index=True)
32	status: Literal["pending", "in_progress", "completed"] = "pending"
33
34	class Task(TaskBase, table=True):
35	id: Optional[int] = Field(default=None, primary_key=True)
36	created_at: datetime = Field(default_factory=datetime.now)
37
38	class TaskCreate(TaskBase):
39	pass
40
41	class TaskRead(TaskBase):
42	id: int
43	created_at: datetime
44	```
45
46	### 2. Async Database Setup
47
48	```python
49	from sqlmodel.ext.asyncio.session import AsyncSession
50	from sqlalchemy.ext.asyncio import create_async_engine
51	import os
52
53	DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
54	engine = create_async_engine(DATABASE_URL)
55
56	async def get_session() -> AsyncSession:
57	async with AsyncSession(engine) as session:
58	yield session
59	```
60
61	### 3. CRUD Endpoints
62
63	```python
64	from fastapi import FastAPI, Depends, HTTPException
65	from sqlmodel import select
66
67	app = FastAPI()
68
69	@app.post("/tasks", response_model=TaskRead, status_code=201)
70	async def create_task(task: TaskCreate, session: AsyncSession = Depends(get_session)):
71	db_task = Task.model_validate(task)
72	session.add(db_task)
73	await session.commit()
74	await session.refresh(db_task)
75	return db_task
76
77	@app.get("/tasks/{task_id}", response_model=TaskRead)
78	async def get_task(task_id: int, session: AsyncSession = Depends(get_session)):
79	task = await session.get(Task, task_id)
80	if not task:
81	raise HTTPException(status_code=404, detail="Not found")
82	return task
83
84	@app.patch("/tasks/{task_id}", response_model=TaskRead)
85	async def update_task(task_id: int, update: TaskUpdate, session: AsyncSession = Depends(get_session)):
86	task = await session.get(Task, task_id)
87	if not task:
88	raise HTTPException(status_code=404, detail="Not found")
89	update_data = update.model_dump(exclude_unset=True)
90	task.sqlmodel_update(update_data)
91	session.add(task)
92	await session.commit()
93	await session.refresh(task)
94	return task
95	```
96
97	### 4. JWT/JWKS Authentication
98
99	```python
100	from jose import jwt
101	import httpx
102
103	JWKS_URL = f"{SSO_URL}/.well-known/jwks.json"
104
105	async def get_current_user(authorization: str = Header()):
106	token = authorization.replace("Bearer ", "")
107	async with httpx.AsyncClient() as client:
108	jwks = (await client.get(JWKS_URL)).json()
109	payload = jwt.decode(token, jwks, algorithms=["RS256"])
110	return payload
111
112	@app.get("/protected")
113	async def protected_route(user = Depends(get_current_user)):
114	return {"user": user["sub"]}
115	```
116
117	See [references/fastapi-patterns.md](references/fastapi-patterns.md) for audit logging, pagination, and OpenAPI configuration.
118
119	---
120
121	## Dapr Integration Patterns
122
123	### 1. Pub/Sub Subscription
124
125	```python
126	from fastapi import APIRouter, Request
127
128	router = APIRouter(prefix="/dapr", tags=["Dapr"])
129
130	@router.get("/subscribe")
131	async def subscribe():
132	"""Dapr calls this to discover subscriptions."""
133	return [{
134	"pubsubname": "pubsub",
135	"topic": "task-created",
136	"route": "/dapr/task-created"
137	}]
138
139	@router.post("/task-created")
140	async def handle_task_created(request: Request, session: AsyncSession = Depends(get_session)):
141	# CloudEvent wrapper - data is nested
142	event = await request.json()
143	task_data = event.get("data", event) # Handle both wrapped and unwrapped
144
145	# Process event
146	task = Task.model_validate(task_data)
147	session.add(task)
148	await session.commit()
149	return {"status": "processed"}
150	```
151
152	### 2. Publishing Events
153
154	```python
155	import httpx
156
157	DAPR_URL = "http://localhost:3500"
158
159	async def publish_event(topic: str, data: dict):
160	async with httpx.AsyncClient() as client:
161	await client.post(
162	f"{DAPR_URL}/v1.0/publish/pubsub/{topic}",
163	json=data,
164	headers={"Content-Type": "application/json"}
165	)
166	```
167
168	### 3. Scheduled Jobs
169
170	```python
171	# Schedule a job via Dapr Jobs API (alpha)
172	async def schedule_job(name: str, schedule: str, callback_url: str, data: dict):

People who liked this also viewed

6 picks

Type