feat(bundles): add editorial bundle plugins
This commit is contained in:
sickn33
@@ -0,0 +1,42 @@
|
||||
---
|
||||
name: async-python-patterns
|
||||
description: "Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems."
|
||||
risk: unknown
|
||||
source: community
|
||||
date_added: "2026-02-27"
|
||||
---
|
||||
|
||||
# Async Python Patterns
|
||||
|
||||
Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems.
|
||||
|
||||
## Use this skill when
|
||||
|
||||
- Building async web APIs (FastAPI, aiohttp, Sanic)
|
||||
- Implementing concurrent I/O operations (database, file, network)
|
||||
- Creating web scrapers with concurrent requests
|
||||
- Developing real-time applications (WebSocket servers, chat systems)
|
||||
- Processing multiple independent tasks simultaneously
|
||||
- Building microservices with async communication
|
||||
- Optimizing I/O-bound workloads
|
||||
- Implementing async background tasks and queues
|
||||
|
||||
## Do not use this skill when
|
||||
|
||||
- The workload is CPU-bound with minimal I/O.
|
||||
- A simple synchronous script is sufficient.
|
||||
- The runtime environment cannot support asyncio/event loop usage.
|
||||
|
||||
## Instructions
|
||||
|
||||
- Clarify workload characteristics (I/O vs CPU), targets, and runtime constraints.
|
||||
- Pick concurrency patterns (tasks, gather, queues, pools) with cancellation rules.
|
||||
- Add timeouts, backpressure, and structured error handling.
|
||||
- Include testing and debugging guidance for async code paths.
|
||||
- If detailed examples are required, open `resources/implementation-playbook.md`.
|
||||
|
||||
Refer to `resources/implementation-playbook.md` for detailed patterns and examples.
|
||||
|
||||
## Resources
|
||||
|
||||
- `resources/implementation-playbook.md` for detailed patterns and examples.
|
||||
@@ -0,0 +1,678 @@
|
||||
# Async Python Patterns Implementation Playbook
|
||||
|
||||
This file contains detailed patterns, checklists, and code samples referenced by the skill.
|
||||
|
||||
## Core Concepts
|
||||
|
||||
### 1. Event Loop
|
||||
The event loop is the heart of asyncio, managing and scheduling asynchronous tasks.
|
||||
|
||||
**Key characteristics:**
|
||||
- Single-threaded cooperative multitasking
|
||||
- Schedules coroutines for execution
|
||||
- Handles I/O operations without blocking
|
||||
- Manages callbacks and futures
|
||||
|
||||
### 2. Coroutines
|
||||
Functions defined with `async def` that can be paused and resumed.
|
||||
|
||||
**Syntax:**
|
||||
```python
|
||||
async def my_coroutine():
|
||||
result = await some_async_operation()
|
||||
return result
|
||||
```
|
||||
|
||||
### 3. Tasks
|
||||
Scheduled coroutines that run concurrently on the event loop.
|
||||
|
||||
### 4. Futures
|
||||
Low-level objects representing eventual results of async operations.
|
||||
|
||||
### 5. Async Context Managers
|
||||
Resources that support `async with` for proper cleanup.
|
||||
|
||||
### 6. Async Iterators
|
||||
Objects that support `async for` for iterating over async data sources.
|
||||
|
||||
## Quick Start
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
|
||||
async def main():
|
||||
print("Hello")
|
||||
await asyncio.sleep(1)
|
||||
print("World")
|
||||
|
||||
# Python 3.7+
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
## Fundamental Patterns
|
||||
|
||||
### Pattern 1: Basic Async/Await
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
|
||||
async def fetch_data(url: str) -> dict:
|
||||
"""Fetch data from URL asynchronously."""
|
||||
await asyncio.sleep(1) # Simulate I/O
|
||||
return {"url": url, "data": "result"}
|
||||
|
||||
async def main():
|
||||
result = await fetch_data("https://api.example.com")
|
||||
print(result)
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
### Pattern 2: Concurrent Execution with gather()
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from typing import List
|
||||
|
||||
async def fetch_user(user_id: int) -> dict:
|
||||
"""Fetch user data."""
|
||||
await asyncio.sleep(0.5)
|
||||
return {"id": user_id, "name": f"User {user_id}"}
|
||||
|
||||
async def fetch_all_users(user_ids: List[int]) -> List[dict]:
|
||||
"""Fetch multiple users concurrently."""
|
||||
tasks = [fetch_user(uid) for uid in user_ids]
|
||||
results = await asyncio.gather(*tasks)
|
||||
return results
|
||||
|
||||
async def main():
|
||||
user_ids = [1, 2, 3, 4, 5]
|
||||
users = await fetch_all_users(user_ids)
|
||||
print(f"Fetched {len(users)} users")
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
### Pattern 3: Task Creation and Management
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
|
||||
async def background_task(name: str, delay: int):
|
||||
"""Long-running background task."""
|
||||
print(f"{name} started")
|
||||
await asyncio.sleep(delay)
|
||||
print(f"{name} completed")
|
||||
return f"Result from {name}"
|
||||
|
||||
async def main():
|
||||
# Create tasks
|
||||
task1 = asyncio.create_task(background_task("Task 1", 2))
|
||||
task2 = asyncio.create_task(background_task("Task 2", 1))
|
||||
|
||||
# Do other work
|
||||
print("Main: doing other work")
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
# Wait for tasks
|
||||
result1 = await task1
|
||||
result2 = await task2
|
||||
|
||||
print(f"Results: {result1}, {result2}")
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
### Pattern 4: Error Handling in Async Code
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from typing import List, Optional
|
||||
|
||||
async def risky_operation(item_id: int) -> dict:
|
||||
"""Operation that might fail."""
|
||||
await asyncio.sleep(0.1)
|
||||
if item_id % 3 == 0:
|
||||
raise ValueError(f"Item {item_id} failed")
|
||||
return {"id": item_id, "status": "success"}
|
||||
|
||||
async def safe_operation(item_id: int) -> Optional[dict]:
|
||||
"""Wrapper with error handling."""
|
||||
try:
|
||||
return await risky_operation(item_id)
|
||||
except ValueError as e:
|
||||
print(f"Error: {e}")
|
||||
return None
|
||||
|
||||
async def process_items(item_ids: List[int]):
|
||||
"""Process multiple items with error handling."""
|
||||
tasks = [safe_operation(iid) for iid in item_ids]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
# Filter out failures
|
||||
successful = [r for r in results if r is not None and not isinstance(r, Exception)]
|
||||
failed = [r for r in results if isinstance(r, Exception)]
|
||||
|
||||
print(f"Success: {len(successful)}, Failed: {len(failed)}")
|
||||
return successful
|
||||
|
||||
asyncio.run(process_items([1, 2, 3, 4, 5, 6]))
|
||||
```
|
||||
|
||||
### Pattern 5: Timeout Handling
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
|
||||
async def slow_operation(delay: int) -> str:
|
||||
"""Operation that takes time."""
|
||||
await asyncio.sleep(delay)
|
||||
return f"Completed after {delay}s"
|
||||
|
||||
async def with_timeout():
|
||||
"""Execute operation with timeout."""
|
||||
try:
|
||||
result = await asyncio.wait_for(slow_operation(5), timeout=2.0)
|
||||
print(result)
|
||||
except asyncio.TimeoutError:
|
||||
print("Operation timed out")
|
||||
|
||||
asyncio.run(with_timeout())
|
||||
```
|
||||
|
||||
## Advanced Patterns
|
||||
|
||||
### Pattern 6: Async Context Managers
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from typing import Optional
|
||||
|
||||
class AsyncDatabaseConnection:
|
||||
"""Async database connection context manager."""
|
||||
|
||||
def __init__(self, dsn: str):
|
||||
self.dsn = dsn
|
||||
self.connection: Optional[object] = None
|
||||
|
||||
async def __aenter__(self):
|
||||
print("Opening connection")
|
||||
await asyncio.sleep(0.1) # Simulate connection
|
||||
self.connection = {"dsn": self.dsn, "connected": True}
|
||||
return self.connection
|
||||
|
||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||
print("Closing connection")
|
||||
await asyncio.sleep(0.1) # Simulate cleanup
|
||||
self.connection = None
|
||||
|
||||
async def query_database():
|
||||
"""Use async context manager."""
|
||||
async with AsyncDatabaseConnection("postgresql://localhost") as conn:
|
||||
print(f"Using connection: {conn}")
|
||||
await asyncio.sleep(0.2) # Simulate query
|
||||
return {"rows": 10}
|
||||
|
||||
asyncio.run(query_database())
|
||||
```
|
||||
|
||||
### Pattern 7: Async Iterators and Generators
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from typing import AsyncIterator
|
||||
|
||||
async def async_range(start: int, end: int, delay: float = 0.1) -> AsyncIterator[int]:
|
||||
"""Async generator that yields numbers with delay."""
|
||||
for i in range(start, end):
|
||||
await asyncio.sleep(delay)
|
||||
yield i
|
||||
|
||||
async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]:
|
||||
"""Fetch paginated data asynchronously."""
|
||||
for page in range(1, max_pages + 1):
|
||||
await asyncio.sleep(0.2) # Simulate API call
|
||||
yield {
|
||||
"page": page,
|
||||
"url": f"{url}?page={page}",
|
||||
"data": [f"item_{page}_{i}" for i in range(5)]
|
||||
}
|
||||
|
||||
async def consume_async_iterator():
|
||||
"""Consume async iterator."""
|
||||
async for number in async_range(1, 5):
|
||||
print(f"Number: {number}")
|
||||
|
||||
print("\nFetching pages:")
|
||||
async for page_data in fetch_pages("https://api.example.com/items", 3):
|
||||
print(f"Page {page_data['page']}: {len(page_data['data'])} items")
|
||||
|
||||
asyncio.run(consume_async_iterator())
|
||||
```
|
||||
|
||||
### Pattern 8: Producer-Consumer Pattern
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from asyncio import Queue
|
||||
from typing import Optional
|
||||
|
||||
async def producer(queue: Queue, producer_id: int, num_items: int):
|
||||
"""Produce items and put them in queue."""
|
||||
for i in range(num_items):
|
||||
item = f"Item-{producer_id}-{i}"
|
||||
await queue.put(item)
|
||||
print(f"Producer {producer_id} produced: {item}")
|
||||
await asyncio.sleep(0.1)
|
||||
await queue.put(None) # Signal completion
|
||||
|
||||
async def consumer(queue: Queue, consumer_id: int):
|
||||
"""Consume items from queue."""
|
||||
while True:
|
||||
item = await queue.get()
|
||||
if item is None:
|
||||
queue.task_done()
|
||||
break
|
||||
|
||||
print(f"Consumer {consumer_id} processing: {item}")
|
||||
await asyncio.sleep(0.2) # Simulate work
|
||||
queue.task_done()
|
||||
|
||||
async def producer_consumer_example():
|
||||
"""Run producer-consumer pattern."""
|
||||
queue = Queue(maxsize=10)
|
||||
|
||||
# Create tasks
|
||||
producers = [
|
||||
asyncio.create_task(producer(queue, i, 5))
|
||||
for i in range(2)
|
||||
]
|
||||
|
||||
consumers = [
|
||||
asyncio.create_task(consumer(queue, i))
|
||||
for i in range(3)
|
||||
]
|
||||
|
||||
# Wait for producers
|
||||
await asyncio.gather(*producers)
|
||||
|
||||
# Wait for queue to be empty
|
||||
await queue.join()
|
||||
|
||||
# Cancel consumers
|
||||
for c in consumers:
|
||||
c.cancel()
|
||||
|
||||
asyncio.run(producer_consumer_example())
|
||||
```
|
||||
|
||||
### Pattern 9: Semaphore for Rate Limiting
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from typing import List
|
||||
|
||||
async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict:
|
||||
"""Make API call with rate limiting."""
|
||||
async with semaphore:
|
||||
print(f"Calling {url}")
|
||||
await asyncio.sleep(0.5) # Simulate API call
|
||||
return {"url": url, "status": 200}
|
||||
|
||||
async def rate_limited_requests(urls: List[str], max_concurrent: int = 5):
|
||||
"""Make multiple requests with rate limiting."""
|
||||
semaphore = asyncio.Semaphore(max_concurrent)
|
||||
tasks = [api_call(url, semaphore) for url in urls]
|
||||
results = await asyncio.gather(*tasks)
|
||||
return results
|
||||
|
||||
async def main():
|
||||
urls = [f"https://api.example.com/item/{i}" for i in range(20)]
|
||||
results = await rate_limited_requests(urls, max_concurrent=3)
|
||||
print(f"Completed {len(results)} requests")
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
### Pattern 10: Async Locks and Synchronization
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
|
||||
class AsyncCounter:
|
||||
"""Thread-safe async counter."""
|
||||
|
||||
def __init__(self):
|
||||
self.value = 0
|
||||
self.lock = asyncio.Lock()
|
||||
|
||||
async def increment(self):
|
||||
"""Safely increment counter."""
|
||||
async with self.lock:
|
||||
current = self.value
|
||||
await asyncio.sleep(0.01) # Simulate work
|
||||
self.value = current + 1
|
||||
|
||||
async def get_value(self) -> int:
|
||||
"""Get current value."""
|
||||
async with self.lock:
|
||||
return self.value
|
||||
|
||||
async def worker(counter: AsyncCounter, worker_id: int):
|
||||
"""Worker that increments counter."""
|
||||
for _ in range(10):
|
||||
await counter.increment()
|
||||
print(f"Worker {worker_id} incremented")
|
||||
|
||||
async def test_counter():
|
||||
"""Test concurrent counter."""
|
||||
counter = AsyncCounter()
|
||||
|
||||
workers = [asyncio.create_task(worker(counter, i)) for i in range(5)]
|
||||
await asyncio.gather(*workers)
|
||||
|
||||
final_value = await counter.get_value()
|
||||
print(f"Final counter value: {final_value}")
|
||||
|
||||
asyncio.run(test_counter())
|
||||
```
|
||||
|
||||
## Real-World Applications
|
||||
|
||||
### Web Scraping with aiohttp
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
import aiohttp
|
||||
from typing import List, Dict
|
||||
|
||||
async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict:
|
||||
"""Fetch single URL."""
|
||||
try:
|
||||
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
|
||||
text = await response.text()
|
||||
return {
|
||||
"url": url,
|
||||
"status": response.status,
|
||||
"length": len(text)
|
||||
}
|
||||
except Exception as e:
|
||||
return {"url": url, "error": str(e)}
|
||||
|
||||
async def scrape_urls(urls: List[str]) -> List[Dict]:
|
||||
"""Scrape multiple URLs concurrently."""
|
||||
async with aiohttp.ClientSession() as session:
|
||||
tasks = [fetch_url(session, url) for url in urls]
|
||||
results = await asyncio.gather(*tasks)
|
||||
return results
|
||||
|
||||
async def main():
|
||||
urls = [
|
||||
"https://httpbin.org/delay/1",
|
||||
"https://httpbin.org/delay/2",
|
||||
"https://httpbin.org/status/404",
|
||||
]
|
||||
|
||||
results = await scrape_urls(urls)
|
||||
for result in results:
|
||||
print(result)
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
### Async Database Operations
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from typing import List, Optional
|
||||
|
||||
# Simulated async database client
|
||||
class AsyncDB:
|
||||
"""Simulated async database."""
|
||||
|
||||
async def execute(self, query: str) -> List[dict]:
|
||||
"""Execute query."""
|
||||
await asyncio.sleep(0.1)
|
||||
return [{"id": 1, "name": "Example"}]
|
||||
|
||||
async def fetch_one(self, query: str) -> Optional[dict]:
|
||||
"""Fetch single row."""
|
||||
await asyncio.sleep(0.1)
|
||||
return {"id": 1, "name": "Example"}
|
||||
|
||||
async def get_user_data(db: AsyncDB, user_id: int) -> dict:
|
||||
"""Fetch user and related data concurrently."""
|
||||
user_task = db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}")
|
||||
orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}")
|
||||
profile_task = db.fetch_one(f"SELECT * FROM profiles WHERE user_id = {user_id}")
|
||||
|
||||
user, orders, profile = await asyncio.gather(user_task, orders_task, profile_task)
|
||||
|
||||
return {
|
||||
"user": user,
|
||||
"orders": orders,
|
||||
"profile": profile
|
||||
}
|
||||
|
||||
async def main():
|
||||
db = AsyncDB()
|
||||
user_data = await get_user_data(db, 1)
|
||||
print(user_data)
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
### WebSocket Server
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
from typing import Set
|
||||
|
||||
# Simulated WebSocket connection
|
||||
class WebSocket:
|
||||
"""Simulated WebSocket."""
|
||||
|
||||
def __init__(self, client_id: str):
|
||||
self.client_id = client_id
|
||||
|
||||
async def send(self, message: str):
|
||||
"""Send message."""
|
||||
print(f"Sending to {self.client_id}: {message}")
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
async def recv(self) -> str:
|
||||
"""Receive message."""
|
||||
await asyncio.sleep(1)
|
||||
return f"Message from {self.client_id}"
|
||||
|
||||
class WebSocketServer:
|
||||
"""Simple WebSocket server."""
|
||||
|
||||
def __init__(self):
|
||||
self.clients: Set[WebSocket] = set()
|
||||
|
||||
async def register(self, websocket: WebSocket):
|
||||
"""Register new client."""
|
||||
self.clients.add(websocket)
|
||||
print(f"Client {websocket.client_id} connected")
|
||||
|
||||
async def unregister(self, websocket: WebSocket):
|
||||
"""Unregister client."""
|
||||
self.clients.remove(websocket)
|
||||
print(f"Client {websocket.client_id} disconnected")
|
||||
|
||||
async def broadcast(self, message: str):
|
||||
"""Broadcast message to all clients."""
|
||||
if self.clients:
|
||||
tasks = [client.send(message) for client in self.clients]
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
async def handle_client(self, websocket: WebSocket):
|
||||
"""Handle individual client connection."""
|
||||
await self.register(websocket)
|
||||
try:
|
||||
async for message in self.message_iterator(websocket):
|
||||
await self.broadcast(f"{websocket.client_id}: {message}")
|
||||
finally:
|
||||
await self.unregister(websocket)
|
||||
|
||||
async def message_iterator(self, websocket: WebSocket):
|
||||
"""Iterate over messages from client."""
|
||||
for _ in range(3): # Simulate 3 messages
|
||||
yield await websocket.recv()
|
||||
```
|
||||
|
||||
## Performance Best Practices
|
||||
|
||||
### 1. Use Connection Pools
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
import aiohttp
|
||||
|
||||
async def with_connection_pool():
|
||||
"""Use connection pool for efficiency."""
|
||||
connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)
|
||||
|
||||
async with aiohttp.ClientSession(connector=connector) as session:
|
||||
tasks = [session.get(f"https://api.example.com/item/{i}") for i in range(50)]
|
||||
responses = await asyncio.gather(*tasks)
|
||||
return responses
|
||||
```
|
||||
|
||||
### 2. Batch Operations
|
||||
|
||||
```python
|
||||
async def batch_process(items: List[str], batch_size: int = 10):
|
||||
"""Process items in batches."""
|
||||
for i in range(0, len(items), batch_size):
|
||||
batch = items[i:i + batch_size]
|
||||
tasks = [process_item(item) for item in batch]
|
||||
await asyncio.gather(*tasks)
|
||||
print(f"Processed batch {i // batch_size + 1}")
|
||||
|
||||
async def process_item(item: str):
|
||||
"""Process single item."""
|
||||
await asyncio.sleep(0.1)
|
||||
return f"Processed: {item}"
|
||||
```
|
||||
|
||||
### 3. Avoid Blocking Operations
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
import concurrent.futures
|
||||
from typing import Any
|
||||
|
||||
def blocking_operation(data: Any) -> Any:
|
||||
"""CPU-intensive blocking operation."""
|
||||
import time
|
||||
time.sleep(1)
|
||||
return data * 2
|
||||
|
||||
async def run_in_executor(data: Any) -> Any:
|
||||
"""Run blocking operation in thread pool."""
|
||||
loop = asyncio.get_event_loop()
|
||||
with concurrent.futures.ThreadPoolExecutor() as pool:
|
||||
result = await loop.run_in_executor(pool, blocking_operation, data)
|
||||
return result
|
||||
|
||||
async def main():
|
||||
results = await asyncio.gather(*[run_in_executor(i) for i in range(5)])
|
||||
print(results)
|
||||
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
## Common Pitfalls
|
||||
|
||||
### 1. Forgetting await
|
||||
|
||||
```python
|
||||
# Wrong - returns coroutine object, doesn't execute
|
||||
result = async_function()
|
||||
|
||||
# Correct
|
||||
result = await async_function()
|
||||
```
|
||||
|
||||
### 2. Blocking the Event Loop
|
||||
|
||||
```python
|
||||
# Wrong - blocks event loop
|
||||
import time
|
||||
async def bad():
|
||||
time.sleep(1) # Blocks!
|
||||
|
||||
# Correct
|
||||
async def good():
|
||||
await asyncio.sleep(1) # Non-blocking
|
||||
```
|
||||
|
||||
### 3. Not Handling Cancellation
|
||||
|
||||
```python
|
||||
async def cancelable_task():
|
||||
"""Task that handles cancellation."""
|
||||
try:
|
||||
while True:
|
||||
await asyncio.sleep(1)
|
||||
print("Working...")
|
||||
except asyncio.CancelledError:
|
||||
print("Task cancelled, cleaning up...")
|
||||
# Perform cleanup
|
||||
raise # Re-raise to propagate cancellation
|
||||
```
|
||||
|
||||
### 4. Mixing Sync and Async Code
|
||||
|
||||
```python
|
||||
# Wrong - can't call async from sync directly
|
||||
def sync_function():
|
||||
result = await async_function() # SyntaxError!
|
||||
|
||||
# Correct
|
||||
def sync_function():
|
||||
result = asyncio.run(async_function())
|
||||
```
|
||||
|
||||
## Testing Async Code
|
||||
|
||||
```python
|
||||
import asyncio
|
||||
import pytest
|
||||
|
||||
# Using pytest-asyncio
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_function():
|
||||
"""Test async function."""
|
||||
result = await fetch_data("https://api.example.com")
|
||||
assert result is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_with_timeout():
|
||||
"""Test with timeout."""
|
||||
with pytest.raises(asyncio.TimeoutError):
|
||||
await asyncio.wait_for(slow_operation(5), timeout=1.0)
|
||||
```
|
||||
|
||||
## Resources
|
||||
|
||||
- **Python asyncio documentation**: https://docs.python.org/3/library/asyncio.html
|
||||
- **aiohttp**: Async HTTP client/server
|
||||
- **FastAPI**: Modern async web framework
|
||||
- **asyncpg**: Async PostgreSQL driver
|
||||
- **motor**: Async MongoDB driver
|
||||
|
||||
## Best Practices Summary
|
||||
|
||||
1. **Use asyncio.run()** for entry point (Python 3.7+)
|
||||
2. **Always await coroutines** to execute them
|
||||
3. **Use gather() for concurrent execution** of multiple tasks
|
||||
4. **Implement proper error handling** with try/except
|
||||
5. **Use timeouts** to prevent hanging operations
|
||||
6. **Pool connections** for better performance
|
||||
7. **Avoid blocking operations** in async code
|
||||
8. **Use semaphores** for rate limiting
|
||||
9. **Handle task cancellation** properly
|
||||
10. **Test async code** with pytest-asyncio
|
||||
178
plugins/antigravity-bundle-python-pro/skills/django-pro/SKILL.md
Normal file
178
plugins/antigravity-bundle-python-pro/skills/django-pro/SKILL.md
Normal file
@@ -0,0 +1,178 @@
|
||||
---
|
||||
name: django-pro
|
||||
description: Master Django 5.x with async views, DRF, Celery, and Django Channels. Build scalable web applications with proper architecture, testing, and deployment.
|
||||
risk: unknown
|
||||
source: community
|
||||
date_added: '2026-02-27'
|
||||
---
|
||||
|
||||
## Use this skill when
|
||||
|
||||
- Working on django pro tasks or workflows
|
||||
- Needing guidance, best practices, or checklists for django pro
|
||||
|
||||
## Do not use this skill when
|
||||
|
||||
- The task is unrelated to django pro
|
||||
- You need a different domain or tool outside this scope
|
||||
|
||||
## Instructions
|
||||
|
||||
- Clarify goals, constraints, and required inputs.
|
||||
- Apply relevant best practices and validate outcomes.
|
||||
- Provide actionable steps and verification.
|
||||
- If detailed examples are required, open `resources/implementation-playbook.md`.
|
||||
|
||||
You are a Django expert specializing in Django 5.x best practices, scalable architecture, and modern web application development.
|
||||
|
||||
## Purpose
|
||||
|
||||
Expert Django developer specializing in Django 5.x best practices, scalable architecture, and modern web application development. Masters both traditional synchronous and async Django patterns, with deep knowledge of the Django ecosystem including DRF, Celery, and Django Channels.
|
||||
|
||||
## Capabilities
|
||||
|
||||
### Core Django Expertise
|
||||
|
||||
- Django 5.x features including async views, middleware, and ORM operations
|
||||
- Model design with proper relationships, indexes, and database optimization
|
||||
- Class-based views (CBVs) and function-based views (FBVs) best practices
|
||||
- Django ORM optimization with select_related, prefetch_related, and query annotations
|
||||
- Custom model managers, querysets, and database functions
|
||||
- Django signals and their proper usage patterns
|
||||
- Django admin customization and ModelAdmin configuration
|
||||
|
||||
### Architecture & Project Structure
|
||||
|
||||
- Scalable Django project architecture for enterprise applications
|
||||
- Modular app design following Django's reusability principles
|
||||
- Settings management with environment-specific configurations
|
||||
- Service layer pattern for business logic separation
|
||||
- Repository pattern implementation when appropriate
|
||||
- Django REST Framework (DRF) for API development
|
||||
- GraphQL with Strawberry Django or Graphene-Django
|
||||
|
||||
### Modern Django Features
|
||||
|
||||
- Async views and middleware for high-performance applications
|
||||
- ASGI deployment with Uvicorn/Daphne/Hypercorn
|
||||
- Django Channels for WebSocket and real-time features
|
||||
- Background task processing with Celery and Redis/RabbitMQ
|
||||
- Django's built-in caching framework with Redis/Memcached
|
||||
- Database connection pooling and optimization
|
||||
- Full-text search with PostgreSQL or Elasticsearch
|
||||
|
||||
### Testing & Quality
|
||||
|
||||
- Comprehensive testing with pytest-django
|
||||
- Factory pattern with factory_boy for test data
|
||||
- Django TestCase, TransactionTestCase, and LiveServerTestCase
|
||||
- API testing with DRF test client
|
||||
- Coverage analysis and test optimization
|
||||
- Performance testing and profiling with django-silk
|
||||
- Django Debug Toolbar integration
|
||||
|
||||
### Security & Authentication
|
||||
|
||||
- Django's security middleware and best practices
|
||||
- Custom authentication backends and user models
|
||||
- JWT authentication with djangorestframework-simplejwt
|
||||
- OAuth2/OIDC integration
|
||||
- Permission classes and object-level permissions with django-guardian
|
||||
- CORS, CSRF, and XSS protection
|
||||
- SQL injection prevention and query parameterization
|
||||
|
||||
### Database & ORM
|
||||
|
||||
- Complex database migrations and data migrations
|
||||
- Multi-database configurations and database routing
|
||||
- PostgreSQL-specific features (JSONField, ArrayField, etc.)
|
||||
- Database performance optimization and query analysis
|
||||
- Raw SQL when necessary with proper parameterization
|
||||
- Database transactions and atomic operations
|
||||
- Connection pooling with django-db-pool or pgbouncer
|
||||
|
||||
### Deployment & DevOps
|
||||
|
||||
- Production-ready Django configurations
|
||||
- Docker containerization with multi-stage builds
|
||||
- Gunicorn/uWSGI configuration for WSGI
|
||||
- Static file serving with WhiteNoise or CDN integration
|
||||
- Media file handling with django-storages
|
||||
- Environment variable management with django-environ
|
||||
- CI/CD pipelines for Django applications
|
||||
|
||||
### Frontend Integration
|
||||
|
||||
- Django templates with modern JavaScript frameworks
|
||||
- HTMX integration for dynamic UIs without complex JavaScript
|
||||
- Django + React/Vue/Angular architectures
|
||||
- Webpack integration with django-webpack-loader
|
||||
- Server-side rendering strategies
|
||||
- API-first development patterns
|
||||
|
||||
### Performance Optimization
|
||||
|
||||
- Database query optimization and indexing strategies
|
||||
- Django ORM query optimization techniques
|
||||
- Caching strategies at multiple levels (query, view, template)
|
||||
- Lazy loading and eager loading patterns
|
||||
- Database connection pooling
|
||||
- Asynchronous task processing
|
||||
- CDN and static file optimization
|
||||
|
||||
### Third-Party Integrations
|
||||
|
||||
- Payment processing (Stripe, PayPal, etc.)
|
||||
- Email backends and transactional email services
|
||||
- SMS and notification services
|
||||
- Cloud storage (AWS S3, Google Cloud Storage, Azure)
|
||||
- Search engines (Elasticsearch, Algolia)
|
||||
- Monitoring and logging (Sentry, DataDog, New Relic)
|
||||
|
||||
## Behavioral Traits
|
||||
|
||||
- Follows Django's "batteries included" philosophy
|
||||
- Emphasizes reusable, maintainable code
|
||||
- Prioritizes security and performance equally
|
||||
- Uses Django's built-in features before reaching for third-party packages
|
||||
- Writes comprehensive tests for all critical paths
|
||||
- Documents code with clear docstrings and type hints
|
||||
- Follows PEP 8 and Django coding style
|
||||
- Implements proper error handling and logging
|
||||
- Considers database implications of all ORM operations
|
||||
- Uses Django's migration system effectively
|
||||
|
||||
## Knowledge Base
|
||||
|
||||
- Django 5.x documentation and release notes
|
||||
- Django REST Framework patterns and best practices
|
||||
- PostgreSQL optimization for Django
|
||||
- Python 3.11+ features and type hints
|
||||
- Modern deployment strategies for Django
|
||||
- Django security best practices and OWASP guidelines
|
||||
- Celery and distributed task processing
|
||||
- Redis for caching and message queuing
|
||||
- Docker and container orchestration
|
||||
- Modern frontend integration patterns
|
||||
|
||||
## Response Approach
|
||||
|
||||
1. **Analyze requirements** for Django-specific considerations
|
||||
2. **Suggest Django-idiomatic solutions** using built-in features
|
||||
3. **Provide production-ready code** with proper error handling
|
||||
4. **Include tests** for the implemented functionality
|
||||
5. **Consider performance implications** of database queries
|
||||
6. **Document security considerations** when relevant
|
||||
7. **Offer migration strategies** for database changes
|
||||
8. **Suggest deployment configurations** when applicable
|
||||
|
||||
## Example Interactions
|
||||
|
||||
- "Help me optimize this Django queryset that's causing N+1 queries"
|
||||
- "Design a scalable Django architecture for a multi-tenant SaaS application"
|
||||
- "Implement async views for handling long-running API requests"
|
||||
- "Create a custom Django admin interface with inline formsets"
|
||||
- "Set up Django Channels for real-time notifications"
|
||||
- "Optimize database queries for a high-traffic Django application"
|
||||
- "Implement JWT authentication with refresh tokens in DRF"
|
||||
- "Create a robust background task system with Celery"
|
||||
@@ -0,0 +1,190 @@
|
||||
---
|
||||
name: fastapi-pro
|
||||
description: Build high-performance async APIs with FastAPI, SQLAlchemy 2.0, and Pydantic V2. Master microservices, WebSockets, and modern Python async patterns.
|
||||
risk: unknown
|
||||
source: community
|
||||
date_added: '2026-02-27'
|
||||
---
|
||||
|
||||
## Use this skill when
|
||||
|
||||
- Working on fastapi pro tasks or workflows
|
||||
- Needing guidance, best practices, or checklists for fastapi pro
|
||||
|
||||
## Do not use this skill when
|
||||
|
||||
- The task is unrelated to fastapi pro
|
||||
- You need a different domain or tool outside this scope
|
||||
|
||||
## Instructions
|
||||
|
||||
- Clarify goals, constraints, and required inputs.
|
||||
- Apply relevant best practices and validate outcomes.
|
||||
- Provide actionable steps and verification.
|
||||
- If detailed examples are required, open `resources/implementation-playbook.md`.
|
||||
|
||||
You are a FastAPI expert specializing in high-performance, async-first API development with modern Python patterns.
|
||||
|
||||
## Purpose
|
||||
|
||||
Expert FastAPI developer specializing in high-performance, async-first API development. Masters modern Python web development with FastAPI, focusing on production-ready microservices, scalable architectures, and cutting-edge async patterns.
|
||||
|
||||
## Capabilities
|
||||
|
||||
### Core FastAPI Expertise
|
||||
|
||||
- FastAPI 0.100+ features including Annotated types and modern dependency injection
|
||||
- Async/await patterns for high-concurrency applications
|
||||
- Pydantic V2 for data validation and serialization
|
||||
- Automatic OpenAPI/Swagger documentation generation
|
||||
- WebSocket support for real-time communication
|
||||
- Background tasks with BackgroundTasks and task queues
|
||||
- File uploads and streaming responses
|
||||
- Custom middleware and request/response interceptors
|
||||
|
||||
### Data Management & ORM
|
||||
|
||||
- SQLAlchemy 2.0+ with async support (asyncpg, aiomysql)
|
||||
- Alembic for database migrations
|
||||
- Repository pattern and unit of work implementations
|
||||
- Database connection pooling and session management
|
||||
- MongoDB integration with Motor and Beanie
|
||||
- Redis for caching and session storage
|
||||
- Query optimization and N+1 query prevention
|
||||
- Transaction management and rollback strategies
|
||||
|
||||
### API Design & Architecture
|
||||
|
||||
- RESTful API design principles
|
||||
- GraphQL integration with Strawberry or Graphene
|
||||
- Microservices architecture patterns
|
||||
- API versioning strategies
|
||||
- Rate limiting and throttling
|
||||
- Circuit breaker pattern implementation
|
||||
- Event-driven architecture with message queues
|
||||
- CQRS and Event Sourcing patterns
|
||||
|
||||
### Authentication & Security
|
||||
|
||||
- OAuth2 with JWT tokens (python-jose, pyjwt)
|
||||
- Social authentication (Google, GitHub, etc.)
|
||||
- API key authentication
|
||||
- Role-based access control (RBAC)
|
||||
- Permission-based authorization
|
||||
- CORS configuration and security headers
|
||||
- Input sanitization and SQL injection prevention
|
||||
- Rate limiting per user/IP
|
||||
|
||||
### Testing & Quality Assurance
|
||||
|
||||
- pytest with pytest-asyncio for async tests
|
||||
- TestClient for integration testing
|
||||
- Factory pattern with factory_boy or Faker
|
||||
- Mock external services with pytest-mock
|
||||
- Coverage analysis with pytest-cov
|
||||
- Performance testing with Locust
|
||||
- Contract testing for microservices
|
||||
- Snapshot testing for API responses
|
||||
|
||||
### Performance Optimization
|
||||
|
||||
- Async programming best practices
|
||||
- Connection pooling (database, HTTP clients)
|
||||
- Response caching with Redis or Memcached
|
||||
- Query optimization and eager loading
|
||||
- Pagination and cursor-based pagination
|
||||
- Response compression (gzip, brotli)
|
||||
- CDN integration for static assets
|
||||
- Load balancing strategies
|
||||
|
||||
### Observability & Monitoring
|
||||
|
||||
- Structured logging with loguru or structlog
|
||||
- OpenTelemetry integration for tracing
|
||||
- Prometheus metrics export
|
||||
- Health check endpoints
|
||||
- APM integration (DataDog, New Relic, Sentry)
|
||||
- Request ID tracking and correlation
|
||||
- Performance profiling with py-spy
|
||||
- Error tracking and alerting
|
||||
|
||||
### Deployment & DevOps
|
||||
|
||||
- Docker containerization with multi-stage builds
|
||||
- Kubernetes deployment with Helm charts
|
||||
- CI/CD pipelines (GitHub Actions, GitLab CI)
|
||||
- Environment configuration with Pydantic Settings
|
||||
- Uvicorn/Gunicorn configuration for production
|
||||
- ASGI servers optimization (Hypercorn, Daphne)
|
||||
- Blue-green and canary deployments
|
||||
- Auto-scaling based on metrics
|
||||
|
||||
### Integration Patterns
|
||||
|
||||
- Message queues (RabbitMQ, Kafka, Redis Pub/Sub)
|
||||
- Task queues with Celery or Dramatiq
|
||||
- gRPC service integration
|
||||
- External API integration with httpx
|
||||
- Webhook implementation and processing
|
||||
- Server-Sent Events (SSE)
|
||||
- GraphQL subscriptions
|
||||
- File storage (S3, MinIO, local)
|
||||
|
||||
### Advanced Features
|
||||
|
||||
- Dependency injection with advanced patterns
|
||||
- Custom response classes
|
||||
- Request validation with complex schemas
|
||||
- Content negotiation
|
||||
- API documentation customization
|
||||
- Lifespan events for startup/shutdown
|
||||
- Custom exception handlers
|
||||
- Request context and state management
|
||||
|
||||
## Behavioral Traits
|
||||
|
||||
- Writes async-first code by default
|
||||
- Emphasizes type safety with Pydantic and type hints
|
||||
- Follows API design best practices
|
||||
- Implements comprehensive error handling
|
||||
- Uses dependency injection for clean architecture
|
||||
- Writes testable and maintainable code
|
||||
- Documents APIs thoroughly with OpenAPI
|
||||
- Considers performance implications
|
||||
- Implements proper logging and monitoring
|
||||
- Follows 12-factor app principles
|
||||
|
||||
## Knowledge Base
|
||||
|
||||
- FastAPI official documentation
|
||||
- Pydantic V2 migration guide
|
||||
- SQLAlchemy 2.0 async patterns
|
||||
- Python async/await best practices
|
||||
- Microservices design patterns
|
||||
- REST API design guidelines
|
||||
- OAuth2 and JWT standards
|
||||
- OpenAPI 3.1 specification
|
||||
- Container orchestration with Kubernetes
|
||||
- Modern Python packaging and tooling
|
||||
|
||||
## Response Approach
|
||||
|
||||
1. **Analyze requirements** for async opportunities
|
||||
2. **Design API contracts** with Pydantic models first
|
||||
3. **Implement endpoints** with proper error handling
|
||||
4. **Add comprehensive validation** using Pydantic
|
||||
5. **Write async tests** covering edge cases
|
||||
6. **Optimize for performance** with caching and pooling
|
||||
7. **Document with OpenAPI** annotations
|
||||
8. **Consider deployment** and scaling strategies
|
||||
|
||||
## Example Interactions
|
||||
|
||||
- "Create a FastAPI microservice with async SQLAlchemy and Redis caching"
|
||||
- "Implement JWT authentication with refresh tokens in FastAPI"
|
||||
- "Design a scalable WebSocket chat system with FastAPI"
|
||||
- "Optimize this FastAPI endpoint that's causing performance issues"
|
||||
- "Set up a complete FastAPI project with Docker and Kubernetes"
|
||||
- "Implement rate limiting and circuit breaker for external API calls"
|
||||
- "Create a GraphQL endpoint alongside REST in FastAPI"
|
||||
- "Build a file upload system with progress tracking"
|
||||
@@ -0,0 +1,35 @@
|
||||
---
|
||||
name: fastapi-templates
|
||||
description: "Create production-ready FastAPI projects with async patterns, dependency injection, and comprehensive error handling. Use when building new FastAPI applications or setting up backend API projects."
|
||||
risk: unknown
|
||||
source: community
|
||||
date_added: "2026-02-27"
|
||||
---
|
||||
|
||||
# FastAPI Project Templates
|
||||
|
||||
Production-ready FastAPI project structures with async patterns, dependency injection, middleware, and best practices for building high-performance APIs.
|
||||
|
||||
## Use this skill when
|
||||
|
||||
- Starting new FastAPI projects from scratch
|
||||
- Implementing async REST APIs with Python
|
||||
- Building high-performance web services and microservices
|
||||
- Creating async applications with PostgreSQL, MongoDB
|
||||
- Setting up API projects with proper structure and testing
|
||||
|
||||
## Do not use this skill when
|
||||
|
||||
- The task is unrelated to fastapi project templates
|
||||
- You need a different domain or tool outside this scope
|
||||
|
||||
## Instructions
|
||||
|
||||
- Clarify goals, constraints, and required inputs.
|
||||
- Apply relevant best practices and validate outcomes.
|
||||
- Provide actionable steps and verification.
|
||||
- If detailed examples are required, open `resources/implementation-playbook.md`.
|
||||
|
||||
## Resources
|
||||
|
||||
- `resources/implementation-playbook.md` for detailed patterns and examples.
|
||||
@@ -0,0 +1,566 @@
|
||||
# FastAPI Project Templates Implementation Playbook
|
||||
|
||||
This file contains detailed patterns, checklists, and code samples referenced by the skill.
|
||||
|
||||
# FastAPI Project Templates
|
||||
|
||||
Production-ready FastAPI project structures with async patterns, dependency injection, middleware, and best practices for building high-performance APIs.
|
||||
|
||||
## When to Use This Skill
|
||||
|
||||
- Starting new FastAPI projects from scratch
|
||||
- Implementing async REST APIs with Python
|
||||
- Building high-performance web services and microservices
|
||||
- Creating async applications with PostgreSQL, MongoDB
|
||||
- Setting up API projects with proper structure and testing
|
||||
|
||||
## Core Concepts
|
||||
|
||||
### 1. Project Structure
|
||||
|
||||
**Recommended Layout:**
|
||||
|
||||
```
|
||||
app/
|
||||
├── api/ # API routes
|
||||
│ ├── v1/
|
||||
│ │ ├── endpoints/
|
||||
│ │ │ ├── users.py
|
||||
│ │ │ ├── auth.py
|
||||
│ │ │ └── items.py
|
||||
│ │ └── router.py
|
||||
│ └── dependencies.py # Shared dependencies
|
||||
├── core/ # Core configuration
|
||||
│ ├── config.py
|
||||
│ ├── security.py
|
||||
│ └── database.py
|
||||
├── models/ # Database models
|
||||
│ ├── user.py
|
||||
│ └── item.py
|
||||
├── schemas/ # Pydantic schemas
|
||||
│ ├── user.py
|
||||
│ └── item.py
|
||||
├── services/ # Business logic
|
||||
│ ├── user_service.py
|
||||
│ └── auth_service.py
|
||||
├── repositories/ # Data access
|
||||
│ ├── user_repository.py
|
||||
│ └── item_repository.py
|
||||
└── main.py # Application entry
|
||||
```
|
||||
|
||||
### 2. Dependency Injection
|
||||
|
||||
FastAPI's built-in DI system using `Depends`:
|
||||
|
||||
- Database session management
|
||||
- Authentication/authorization
|
||||
- Shared business logic
|
||||
- Configuration injection
|
||||
|
||||
### 3. Async Patterns
|
||||
|
||||
Proper async/await usage:
|
||||
|
||||
- Async route handlers
|
||||
- Async database operations
|
||||
- Async background tasks
|
||||
- Async middleware
|
||||
|
||||
## Implementation Patterns
|
||||
|
||||
### Pattern 1: Complete FastAPI Application
|
||||
|
||||
```python
|
||||
# main.py
|
||||
from fastapi import FastAPI, Depends
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""Application lifespan events."""
|
||||
# Startup
|
||||
await database.connect()
|
||||
yield
|
||||
# Shutdown
|
||||
await database.disconnect()
|
||||
|
||||
app = FastAPI(
|
||||
title="API Template",
|
||||
version="1.0.0",
|
||||
lifespan=lifespan
|
||||
)
|
||||
|
||||
# CORS middleware
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# Include routers
|
||||
from app.api.v1.router import api_router
|
||||
app.include_router(api_router, prefix="/api/v1")
|
||||
|
||||
# core/config.py
|
||||
from pydantic_settings import BaseSettings
|
||||
from functools import lru_cache
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""Application settings."""
|
||||
DATABASE_URL: str
|
||||
SECRET_KEY: str
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
|
||||
API_V1_STR: str = "/api/v1"
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
|
||||
@lru_cache()
|
||||
def get_settings() -> Settings:
|
||||
return Settings()
|
||||
|
||||
# core/database.py
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from app.core.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
engine = create_async_engine(
|
||||
settings.DATABASE_URL,
|
||||
echo=True,
|
||||
future=True
|
||||
)
|
||||
|
||||
AsyncSessionLocal = sessionmaker(
|
||||
engine,
|
||||
class_=AsyncSession,
|
||||
expire_on_commit=False
|
||||
)
|
||||
|
||||
Base = declarative_base()
|
||||
|
||||
async def get_db() -> AsyncSession:
|
||||
"""Dependency for database session."""
|
||||
async with AsyncSessionLocal() as session:
|
||||
try:
|
||||
yield session
|
||||
await session.commit()
|
||||
except Exception:
|
||||
await session.rollback()
|
||||
raise
|
||||
finally:
|
||||
await session.close()
|
||||
```
|
||||
|
||||
### Pattern 2: CRUD Repository Pattern
|
||||
|
||||
```python
|
||||
# repositories/base_repository.py
|
||||
from typing import Generic, TypeVar, Type, Optional, List
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy import select
|
||||
from pydantic import BaseModel
|
||||
|
||||
ModelType = TypeVar("ModelType")
|
||||
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
|
||||
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
|
||||
|
||||
class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
|
||||
"""Base repository for CRUD operations."""
|
||||
|
||||
def __init__(self, model: Type[ModelType]):
|
||||
self.model = model
|
||||
|
||||
async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
|
||||
"""Get by ID."""
|
||||
result = await db.execute(
|
||||
select(self.model).where(self.model.id == id)
|
||||
)
|
||||
return result.scalars().first()
|
||||
|
||||
async def get_multi(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
skip: int = 0,
|
||||
limit: int = 100
|
||||
) -> List[ModelType]:
|
||||
"""Get multiple records."""
|
||||
result = await db.execute(
|
||||
select(self.model).offset(skip).limit(limit)
|
||||
)
|
||||
return result.scalars().all()
|
||||
|
||||
async def create(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
obj_in: CreateSchemaType
|
||||
) -> ModelType:
|
||||
"""Create new record."""
|
||||
db_obj = self.model(**obj_in.dict())
|
||||
db.add(db_obj)
|
||||
await db.flush()
|
||||
await db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
async def update(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
db_obj: ModelType,
|
||||
obj_in: UpdateSchemaType
|
||||
) -> ModelType:
|
||||
"""Update record."""
|
||||
update_data = obj_in.dict(exclude_unset=True)
|
||||
for field, value in update_data.items():
|
||||
setattr(db_obj, field, value)
|
||||
await db.flush()
|
||||
await db.refresh(db_obj)
|
||||
return db_obj
|
||||
|
||||
async def delete(self, db: AsyncSession, id: int) -> bool:
|
||||
"""Delete record."""
|
||||
obj = await self.get(db, id)
|
||||
if obj:
|
||||
await db.delete(obj)
|
||||
return True
|
||||
return False
|
||||
|
||||
# repositories/user_repository.py
|
||||
from app.repositories.base_repository import BaseRepository
|
||||
from app.models.user import User
|
||||
from app.schemas.user import UserCreate, UserUpdate
|
||||
|
||||
class UserRepository(BaseRepository[User, UserCreate, UserUpdate]):
|
||||
"""User-specific repository."""
|
||||
|
||||
async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]:
|
||||
"""Get user by email."""
|
||||
result = await db.execute(
|
||||
select(User).where(User.email == email)
|
||||
)
|
||||
return result.scalars().first()
|
||||
|
||||
async def is_active(self, db: AsyncSession, user_id: int) -> bool:
|
||||
"""Check if user is active."""
|
||||
user = await self.get(db, user_id)
|
||||
return user.is_active if user else False
|
||||
|
||||
user_repository = UserRepository(User)
|
||||
```
|
||||
|
||||
### Pattern 3: Service Layer
|
||||
|
||||
```python
|
||||
# services/user_service.py
|
||||
from typing import Optional
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from app.repositories.user_repository import user_repository
|
||||
from app.schemas.user import UserCreate, UserUpdate, User
|
||||
from app.core.security import get_password_hash, verify_password
|
||||
|
||||
class UserService:
|
||||
"""Business logic for users."""
|
||||
|
||||
def __init__(self):
|
||||
self.repository = user_repository
|
||||
|
||||
async def create_user(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
user_in: UserCreate
|
||||
) -> User:
|
||||
"""Create new user with hashed password."""
|
||||
# Check if email exists
|
||||
existing = await self.repository.get_by_email(db, user_in.email)
|
||||
if existing:
|
||||
raise ValueError("Email already registered")
|
||||
|
||||
# Hash password
|
||||
user_in_dict = user_in.dict()
|
||||
user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password"))
|
||||
|
||||
# Create user
|
||||
user = await self.repository.create(db, UserCreate(**user_in_dict))
|
||||
return user
|
||||
|
||||
async def authenticate(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
email: str,
|
||||
password: str
|
||||
) -> Optional[User]:
|
||||
"""Authenticate user."""
|
||||
user = await self.repository.get_by_email(db, email)
|
||||
if not user:
|
||||
return None
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return None
|
||||
return user
|
||||
|
||||
async def update_user(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
user_id: int,
|
||||
user_in: UserUpdate
|
||||
) -> Optional[User]:
|
||||
"""Update user."""
|
||||
user = await self.repository.get(db, user_id)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
if user_in.password:
|
||||
user_in_dict = user_in.dict(exclude_unset=True)
|
||||
user_in_dict["hashed_password"] = get_password_hash(
|
||||
user_in_dict.pop("password")
|
||||
)
|
||||
user_in = UserUpdate(**user_in_dict)
|
||||
|
||||
return await self.repository.update(db, user, user_in)
|
||||
|
||||
user_service = UserService()
|
||||
```
|
||||
|
||||
### Pattern 4: API Endpoints with Dependencies
|
||||
|
||||
```python
|
||||
# api/v1/endpoints/users.py
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from typing import List
|
||||
|
||||
from app.core.database import get_db
|
||||
from app.schemas.user import User, UserCreate, UserUpdate
|
||||
from app.services.user_service import user_service
|
||||
from app.api.dependencies import get_current_user
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
|
||||
async def create_user(
|
||||
user_in: UserCreate,
|
||||
db: AsyncSession = Depends(get_db)
|
||||
):
|
||||
"""Create new user."""
|
||||
try:
|
||||
user = await user_service.create_user(db, user_in)
|
||||
return user
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
|
||||
@router.get("/me", response_model=User)
|
||||
async def read_current_user(
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Get current user."""
|
||||
return current_user
|
||||
|
||||
@router.get("/{user_id}", response_model=User)
|
||||
async def read_user(
|
||||
user_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Get user by ID."""
|
||||
user = await user_service.repository.get(db, user_id)
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return user
|
||||
|
||||
@router.patch("/{user_id}", response_model=User)
|
||||
async def update_user(
|
||||
user_id: int,
|
||||
user_in: UserUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Update user."""
|
||||
if current_user.id != user_id:
|
||||
raise HTTPException(status_code=403, detail="Not authorized")
|
||||
|
||||
user = await user_service.update_user(db, user_id, user_in)
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return user
|
||||
|
||||
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_user(
|
||||
user_id: int,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(get_current_user)
|
||||
):
|
||||
"""Delete user."""
|
||||
if current_user.id != user_id:
|
||||
raise HTTPException(status_code=403, detail="Not authorized")
|
||||
|
||||
deleted = await user_service.repository.delete(db, user_id)
|
||||
if not deleted:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
```
|
||||
|
||||
### Pattern 5: Authentication & Authorization
|
||||
|
||||
```python
|
||||
# core/security.py
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional
|
||||
from jose import JWTError, jwt
|
||||
from passlib.context import CryptContext
|
||||
from app.core.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
ALGORITHM = "HS256"
|
||||
|
||||
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
|
||||
"""Create JWT access token."""
|
||||
to_encode = data.copy()
|
||||
if expires_delta:
|
||||
expire = datetime.utcnow() + expires_delta
|
||||
else:
|
||||
expire = datetime.utcnow() + timedelta(minutes=15)
|
||||
to_encode.update({"exp": expire})
|
||||
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||||
"""Verify password against hash."""
|
||||
return pwd_context.verify(plain_password, hashed_password)
|
||||
|
||||
def get_password_hash(password: str) -> str:
|
||||
"""Hash password."""
|
||||
return pwd_context.hash(password)
|
||||
|
||||
# api/dependencies.py
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from jose import JWTError, jwt
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.database import get_db
|
||||
from app.core.security import ALGORITHM
|
||||
from app.core.config import get_settings
|
||||
from app.repositories.user_repository import user_repository
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
|
||||
|
||||
async def get_current_user(
|
||||
db: AsyncSession = Depends(get_db),
|
||||
token: str = Depends(oauth2_scheme)
|
||||
):
|
||||
"""Get current authenticated user."""
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
|
||||
try:
|
||||
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id: int = payload.get("sub")
|
||||
if user_id is None:
|
||||
raise credentials_exception
|
||||
except JWTError:
|
||||
raise credentials_exception
|
||||
|
||||
user = await user_repository.get(db, user_id)
|
||||
if user is None:
|
||||
raise credentials_exception
|
||||
|
||||
return user
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
```python
|
||||
# tests/conftest.py
|
||||
import pytest
|
||||
import asyncio
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from app.main import app
|
||||
from app.core.database import get_db, Base
|
||||
|
||||
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def event_loop():
|
||||
loop = asyncio.get_event_loop_policy().new_event_loop()
|
||||
yield loop
|
||||
loop.close()
|
||||
|
||||
@pytest.fixture
|
||||
async def db_session():
|
||||
engine = create_async_engine(TEST_DATABASE_URL, echo=True)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
AsyncSessionLocal = sessionmaker(
|
||||
engine, class_=AsyncSession, expire_on_commit=False
|
||||
)
|
||||
|
||||
async with AsyncSessionLocal() as session:
|
||||
yield session
|
||||
|
||||
@pytest.fixture
|
||||
async def client(db_session):
|
||||
async def override_get_db():
|
||||
yield db_session
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
yield client
|
||||
|
||||
# tests/test_users.py
|
||||
import pytest
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_user(client):
|
||||
response = await client.post(
|
||||
"/api/v1/users/",
|
||||
json={
|
||||
"email": "test@example.com",
|
||||
"password": "testpass123",
|
||||
"name": "Test User"
|
||||
}
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["email"] == "test@example.com"
|
||||
assert "id" in data
|
||||
```
|
||||
|
||||
## Resources
|
||||
|
||||
- **references/fastapi-architecture.md**: Detailed architecture guide
|
||||
- **references/async-best-practices.md**: Async/await patterns
|
||||
- **references/testing-strategies.md**: Comprehensive testing guide
|
||||
- **assets/project-template/**: Complete FastAPI project
|
||||
- **assets/docker-compose.yml**: Development environment setup
|
||||
|
||||
## Best Practices
|
||||
|
||||
1. **Async All The Way**: Use async for database, external APIs
|
||||
2. **Dependency Injection**: Leverage FastAPI's DI system
|
||||
3. **Repository Pattern**: Separate data access from business logic
|
||||
4. **Service Layer**: Keep business logic out of routes
|
||||
5. **Pydantic Schemas**: Strong typing for request/response
|
||||
6. **Error Handling**: Consistent error responses
|
||||
7. **Testing**: Test all layers independently
|
||||
|
||||
## Common Pitfalls
|
||||
|
||||
- **Blocking Code in Async**: Using synchronous database drivers
|
||||
- **No Service Layer**: Business logic in route handlers
|
||||
- **Missing Type Hints**: Loses FastAPI's benefits
|
||||
- **Ignoring Sessions**: Not properly managing database sessions
|
||||
- **No Testing**: Skipping integration tests
|
||||
- **Tight Coupling**: Direct database access in routes
|
||||
@@ -0,0 +1,446 @@
|
||||
---
|
||||
name: python-patterns
|
||||
description: "Python development principles and decision-making. Framework selection, async patterns, type hints, project structure. Teaches thinking, not copying."
|
||||
risk: unknown
|
||||
source: community
|
||||
date_added: "2026-02-27"
|
||||
---
|
||||
|
||||
# Python Patterns
|
||||
|
||||
> Python development principles and decision-making for 2025.
|
||||
> **Learn to THINK, not memorize patterns.**
|
||||
|
||||
## When to Use
|
||||
Use this skill when making Python architecture decisions, choosing frameworks, designing async patterns, or structuring Python projects.
|
||||
|
||||
---
|
||||
|
||||
## ⚠️ How to Use This Skill
|
||||
|
||||
This skill teaches **decision-making principles**, not fixed code to copy.
|
||||
|
||||
- ASK user for framework preference when unclear
|
||||
- Choose async vs sync based on CONTEXT
|
||||
- Don't default to same framework every time
|
||||
|
||||
---
|
||||
|
||||
## 1. Framework Selection (2025)
|
||||
|
||||
### Decision Tree
|
||||
|
||||
```
|
||||
What are you building?
|
||||
│
|
||||
├── API-first / Microservices
|
||||
│ └── FastAPI (async, modern, fast)
|
||||
│
|
||||
├── Full-stack web / CMS / Admin
|
||||
│ └── Django (batteries-included)
|
||||
│
|
||||
├── Simple / Script / Learning
|
||||
│ └── Flask (minimal, flexible)
|
||||
│
|
||||
├── AI/ML API serving
|
||||
│ └── FastAPI (Pydantic, async, uvicorn)
|
||||
│
|
||||
└── Background workers
|
||||
└── Celery + any framework
|
||||
```
|
||||
|
||||
### Comparison Principles
|
||||
|
||||
| Factor | FastAPI | Django | Flask |
|
||||
|--------|---------|--------|-------|
|
||||
| **Best for** | APIs, microservices | Full-stack, CMS | Simple, learning |
|
||||
| **Async** | Native | Django 5.0+ | Via extensions |
|
||||
| **Admin** | Manual | Built-in | Via extensions |
|
||||
| **ORM** | Choose your own | Django ORM | Choose your own |
|
||||
| **Learning curve** | Low | Medium | Low |
|
||||
|
||||
### Selection Questions to Ask:
|
||||
1. Is this API-only or full-stack?
|
||||
2. Need admin interface?
|
||||
3. Team familiar with async?
|
||||
4. Existing infrastructure?
|
||||
|
||||
---
|
||||
|
||||
## 2. Async vs Sync Decision
|
||||
|
||||
### When to Use Async
|
||||
|
||||
```
|
||||
async def is better when:
|
||||
├── I/O-bound operations (database, HTTP, file)
|
||||
├── Many concurrent connections
|
||||
├── Real-time features
|
||||
├── Microservices communication
|
||||
└── FastAPI/Starlette/Django ASGI
|
||||
|
||||
def (sync) is better when:
|
||||
├── CPU-bound operations
|
||||
├── Simple scripts
|
||||
├── Legacy codebase
|
||||
├── Team unfamiliar with async
|
||||
└── Blocking libraries (no async version)
|
||||
```
|
||||
|
||||
### The Golden Rule
|
||||
|
||||
```
|
||||
I/O-bound → async (waiting for external)
|
||||
CPU-bound → sync + multiprocessing (computing)
|
||||
|
||||
Don't:
|
||||
├── Mix sync and async carelessly
|
||||
├── Use sync libraries in async code
|
||||
└── Force async for CPU work
|
||||
```
|
||||
|
||||
### Async Library Selection
|
||||
|
||||
| Need | Async Library |
|
||||
|------|---------------|
|
||||
| HTTP client | httpx |
|
||||
| PostgreSQL | asyncpg |
|
||||
| Redis | aioredis / redis-py async |
|
||||
| File I/O | aiofiles |
|
||||
| Database ORM | SQLAlchemy 2.0 async, Tortoise |
|
||||
|
||||
---
|
||||
|
||||
## 3. Type Hints Strategy
|
||||
|
||||
### When to Type
|
||||
|
||||
```
|
||||
Always type:
|
||||
├── Function parameters
|
||||
├── Return types
|
||||
├── Class attributes
|
||||
├── Public APIs
|
||||
|
||||
Can skip:
|
||||
├── Local variables (let inference work)
|
||||
├── One-off scripts
|
||||
├── Tests (usually)
|
||||
```
|
||||
|
||||
### Common Type Patterns
|
||||
|
||||
```python
|
||||
# These are patterns, understand them:
|
||||
|
||||
# Optional → might be None
|
||||
from typing import Optional
|
||||
def find_user(id: int) -> Optional[User]: ...
|
||||
|
||||
# Union → one of multiple types
|
||||
def process(data: str | dict) -> None: ...
|
||||
|
||||
# Generic collections
|
||||
def get_items() -> list[Item]: ...
|
||||
def get_mapping() -> dict[str, int]: ...
|
||||
|
||||
# Callable
|
||||
from typing import Callable
|
||||
def apply(fn: Callable[[int], str]) -> str: ...
|
||||
```
|
||||
|
||||
### Pydantic for Validation
|
||||
|
||||
```
|
||||
When to use Pydantic:
|
||||
├── API request/response models
|
||||
├── Configuration/settings
|
||||
├── Data validation
|
||||
├── Serialization
|
||||
|
||||
Benefits:
|
||||
├── Runtime validation
|
||||
├── Auto-generated JSON schema
|
||||
├── Works with FastAPI natively
|
||||
└── Clear error messages
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 4. Project Structure Principles
|
||||
|
||||
### Structure Selection
|
||||
|
||||
```
|
||||
Small project / Script:
|
||||
├── main.py
|
||||
├── utils.py
|
||||
└── requirements.txt
|
||||
|
||||
Medium API:
|
||||
├── app/
|
||||
│ ├── __init__.py
|
||||
│ ├── main.py
|
||||
│ ├── models/
|
||||
│ ├── routes/
|
||||
│ ├── services/
|
||||
│ └── schemas/
|
||||
├── tests/
|
||||
└── pyproject.toml
|
||||
|
||||
Large application:
|
||||
├── src/
|
||||
│ └── myapp/
|
||||
│ ├── core/
|
||||
│ ├── api/
|
||||
│ ├── services/
|
||||
│ ├── models/
|
||||
│ └── ...
|
||||
├── tests/
|
||||
└── pyproject.toml
|
||||
```
|
||||
|
||||
### FastAPI Structure Principles
|
||||
|
||||
```
|
||||
Organize by feature or layer:
|
||||
|
||||
By layer:
|
||||
├── routes/ (API endpoints)
|
||||
├── services/ (business logic)
|
||||
├── models/ (database models)
|
||||
├── schemas/ (Pydantic models)
|
||||
└── dependencies/ (shared deps)
|
||||
|
||||
By feature:
|
||||
├── users/
|
||||
│ ├── routes.py
|
||||
│ ├── service.py
|
||||
│ └── schemas.py
|
||||
└── products/
|
||||
└── ...
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 5. Django Principles (2025)
|
||||
|
||||
### Django Async (Django 5.0+)
|
||||
|
||||
```
|
||||
Django supports async:
|
||||
├── Async views
|
||||
├── Async middleware
|
||||
├── Async ORM (limited)
|
||||
└── ASGI deployment
|
||||
|
||||
When to use async in Django:
|
||||
├── External API calls
|
||||
├── WebSocket (Channels)
|
||||
├── High-concurrency views
|
||||
└── Background task triggering
|
||||
```
|
||||
|
||||
### Django Best Practices
|
||||
|
||||
```
|
||||
Model design:
|
||||
├── Fat models, thin views
|
||||
├── Use managers for common queries
|
||||
├── Abstract base classes for shared fields
|
||||
|
||||
Views:
|
||||
├── Class-based for complex CRUD
|
||||
├── Function-based for simple endpoints
|
||||
├── Use viewsets with DRF
|
||||
|
||||
Queries:
|
||||
├── select_related() for FKs
|
||||
├── prefetch_related() for M2M
|
||||
├── Avoid N+1 queries
|
||||
└── Use .only() for specific fields
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 6. FastAPI Principles
|
||||
|
||||
### async def vs def in FastAPI
|
||||
|
||||
```
|
||||
Use async def when:
|
||||
├── Using async database drivers
|
||||
├── Making async HTTP calls
|
||||
├── I/O-bound operations
|
||||
└── Want to handle concurrency
|
||||
|
||||
Use def when:
|
||||
├── Blocking operations
|
||||
├── Sync database drivers
|
||||
├── CPU-bound work
|
||||
└── FastAPI runs in threadpool automatically
|
||||
```
|
||||
|
||||
### Dependency Injection
|
||||
|
||||
```
|
||||
Use dependencies for:
|
||||
├── Database sessions
|
||||
├── Current user / Auth
|
||||
├── Configuration
|
||||
├── Shared resources
|
||||
|
||||
Benefits:
|
||||
├── Testability (mock dependencies)
|
||||
├── Clean separation
|
||||
├── Automatic cleanup (yield)
|
||||
```
|
||||
|
||||
### Pydantic v2 Integration
|
||||
|
||||
```python
|
||||
# FastAPI + Pydantic are tightly integrated:
|
||||
|
||||
# Request validation
|
||||
@app.post("/users")
|
||||
async def create(user: UserCreate) -> UserResponse:
|
||||
# user is already validated
|
||||
...
|
||||
|
||||
# Response serialization
|
||||
# Return type becomes response schema
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 7. Background Tasks
|
||||
|
||||
### Selection Guide
|
||||
|
||||
| Solution | Best For |
|
||||
|----------|----------|
|
||||
| **BackgroundTasks** | Simple, in-process tasks |
|
||||
| **Celery** | Distributed, complex workflows |
|
||||
| **ARQ** | Async, Redis-based |
|
||||
| **RQ** | Simple Redis queue |
|
||||
| **Dramatiq** | Actor-based, simpler than Celery |
|
||||
|
||||
### When to Use Each
|
||||
|
||||
```
|
||||
FastAPI BackgroundTasks:
|
||||
├── Quick operations
|
||||
├── No persistence needed
|
||||
├── Fire-and-forget
|
||||
└── Same process
|
||||
|
||||
Celery/ARQ:
|
||||
├── Long-running tasks
|
||||
├── Need retry logic
|
||||
├── Distributed workers
|
||||
├── Persistent queue
|
||||
└── Complex workflows
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 8. Error Handling Principles
|
||||
|
||||
### Exception Strategy
|
||||
|
||||
```
|
||||
In FastAPI:
|
||||
├── Create custom exception classes
|
||||
├── Register exception handlers
|
||||
├── Return consistent error format
|
||||
└── Log without exposing internals
|
||||
|
||||
Pattern:
|
||||
├── Raise domain exceptions in services
|
||||
├── Catch and transform in handlers
|
||||
└── Client gets clean error response
|
||||
```
|
||||
|
||||
### Error Response Philosophy
|
||||
|
||||
```
|
||||
Include:
|
||||
├── Error code (programmatic)
|
||||
├── Message (human readable)
|
||||
├── Details (field-level when applicable)
|
||||
└── NOT stack traces (security)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 9. Testing Principles
|
||||
|
||||
### Testing Strategy
|
||||
|
||||
| Type | Purpose | Tools |
|
||||
|------|---------|-------|
|
||||
| **Unit** | Business logic | pytest |
|
||||
| **Integration** | API endpoints | pytest + httpx/TestClient |
|
||||
| **E2E** | Full workflows | pytest + DB |
|
||||
|
||||
### Async Testing
|
||||
|
||||
```python
|
||||
# Use pytest-asyncio for async tests
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_endpoint():
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
response = await client.get("/users")
|
||||
assert response.status_code == 200
|
||||
```
|
||||
|
||||
### Fixtures Strategy
|
||||
|
||||
```
|
||||
Common fixtures:
|
||||
├── db_session → Database connection
|
||||
├── client → Test client
|
||||
├── authenticated_user → User with token
|
||||
└── sample_data → Test data setup
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 10. Decision Checklist
|
||||
|
||||
Before implementing:
|
||||
|
||||
- [ ] **Asked user about framework preference?**
|
||||
- [ ] **Chosen framework for THIS context?** (not just default)
|
||||
- [ ] **Decided async vs sync?**
|
||||
- [ ] **Planned type hint strategy?**
|
||||
- [ ] **Defined project structure?**
|
||||
- [ ] **Planned error handling?**
|
||||
- [ ] **Considered background tasks?**
|
||||
|
||||
---
|
||||
|
||||
## 11. Anti-Patterns to Avoid
|
||||
|
||||
### ❌ DON'T:
|
||||
- Default to Django for simple APIs (FastAPI may be better)
|
||||
- Use sync libraries in async code
|
||||
- Skip type hints for public APIs
|
||||
- Put business logic in routes/views
|
||||
- Ignore N+1 queries
|
||||
- Mix async and sync carelessly
|
||||
|
||||
### ✅ DO:
|
||||
- Choose framework based on context
|
||||
- Ask about async requirements
|
||||
- Use Pydantic for validation
|
||||
- Separate concerns (routes → services → repos)
|
||||
- Test critical paths
|
||||
|
||||
---
|
||||
|
||||
> **Remember**: Python patterns are about decision-making for YOUR specific context. Don't copy code—think about what serves your application best.
|
||||
156
plugins/antigravity-bundle-python-pro/skills/python-pro/SKILL.md
Normal file
156
plugins/antigravity-bundle-python-pro/skills/python-pro/SKILL.md
Normal file
@@ -0,0 +1,156 @@
|
||||
---
|
||||
name: python-pro
|
||||
description: Master Python 3.12+ with modern features, async programming, performance optimization, and production-ready practices. Expert in the latest Python ecosystem including uv, ruff, pydantic, and FastAPI.
|
||||
risk: unknown
|
||||
source: community
|
||||
date_added: '2026-02-27'
|
||||
---
|
||||
You are a Python expert specializing in modern Python 3.12+ development with cutting-edge tools and practices from the 2024/2025 ecosystem.
|
||||
|
||||
## Use this skill when
|
||||
|
||||
- Writing or reviewing Python 3.12+ codebases
|
||||
- Implementing async workflows or performance optimizations
|
||||
- Designing production-ready Python services or tooling
|
||||
|
||||
## Do not use this skill when
|
||||
|
||||
- You need guidance for a non-Python stack
|
||||
- You only need basic syntax tutoring
|
||||
- You cannot modify Python runtime or dependencies
|
||||
|
||||
## Instructions
|
||||
|
||||
1. Confirm runtime, dependencies, and performance targets.
|
||||
2. Choose patterns (async, typing, tooling) that match requirements.
|
||||
3. Implement and test with modern tooling.
|
||||
4. Profile and tune for latency, memory, and correctness.
|
||||
|
||||
## Purpose
|
||||
Expert Python developer mastering Python 3.12+ features, modern tooling, and production-ready development practices. Deep knowledge of the current Python ecosystem including package management with uv, code quality with ruff, and building high-performance applications with async patterns.
|
||||
|
||||
## Capabilities
|
||||
|
||||
### Modern Python Features
|
||||
- Python 3.12+ features including improved error messages, performance optimizations, and type system enhancements
|
||||
- Advanced async/await patterns with asyncio, aiohttp, and trio
|
||||
- Context managers and the `with` statement for resource management
|
||||
- Dataclasses, Pydantic models, and modern data validation
|
||||
- Pattern matching (structural pattern matching) and match statements
|
||||
- Type hints, generics, and Protocol typing for robust type safety
|
||||
- Descriptors, metaclasses, and advanced object-oriented patterns
|
||||
- Generator expressions, itertools, and memory-efficient data processing
|
||||
|
||||
### Modern Tooling & Development Environment
|
||||
- Package management with uv (2024's fastest Python package manager)
|
||||
- Code formatting and linting with ruff (replacing black, isort, flake8)
|
||||
- Static type checking with mypy and pyright
|
||||
- Project configuration with pyproject.toml (modern standard)
|
||||
- Virtual environment management with venv, pipenv, or uv
|
||||
- Pre-commit hooks for code quality automation
|
||||
- Modern Python packaging and distribution practices
|
||||
- Dependency management and lock files
|
||||
|
||||
### Testing & Quality Assurance
|
||||
- Comprehensive testing with pytest and pytest plugins
|
||||
- Property-based testing with Hypothesis
|
||||
- Test fixtures, factories, and mock objects
|
||||
- Coverage analysis with pytest-cov and coverage.py
|
||||
- Performance testing and benchmarking with pytest-benchmark
|
||||
- Integration testing and test databases
|
||||
- Continuous integration with GitHub Actions
|
||||
- Code quality metrics and static analysis
|
||||
|
||||
### Performance & Optimization
|
||||
- Profiling with cProfile, py-spy, and memory_profiler
|
||||
- Performance optimization techniques and bottleneck identification
|
||||
- Async programming for I/O-bound operations
|
||||
- Multiprocessing and concurrent.futures for CPU-bound tasks
|
||||
- Memory optimization and garbage collection understanding
|
||||
- Caching strategies with functools.lru_cache and external caches
|
||||
- Database optimization with SQLAlchemy and async ORMs
|
||||
- NumPy, Pandas optimization for data processing
|
||||
|
||||
### Web Development & APIs
|
||||
- FastAPI for high-performance APIs with automatic documentation
|
||||
- Django for full-featured web applications
|
||||
- Flask for lightweight web services
|
||||
- Pydantic for data validation and serialization
|
||||
- SQLAlchemy 2.0+ with async support
|
||||
- Background task processing with Celery and Redis
|
||||
- WebSocket support with FastAPI and Django Channels
|
||||
- Authentication and authorization patterns
|
||||
|
||||
### Data Science & Machine Learning
|
||||
- NumPy and Pandas for data manipulation and analysis
|
||||
- Matplotlib, Seaborn, and Plotly for data visualization
|
||||
- Scikit-learn for machine learning workflows
|
||||
- Jupyter notebooks and IPython for interactive development
|
||||
- Data pipeline design and ETL processes
|
||||
- Integration with modern ML libraries (PyTorch, TensorFlow)
|
||||
- Data validation and quality assurance
|
||||
- Performance optimization for large datasets
|
||||
|
||||
### DevOps & Production Deployment
|
||||
- Docker containerization and multi-stage builds
|
||||
- Kubernetes deployment and scaling strategies
|
||||
- Cloud deployment (AWS, GCP, Azure) with Python services
|
||||
- Monitoring and logging with structured logging and APM tools
|
||||
- Configuration management and environment variables
|
||||
- Security best practices and vulnerability scanning
|
||||
- CI/CD pipelines and automated testing
|
||||
- Performance monitoring and alerting
|
||||
|
||||
### Advanced Python Patterns
|
||||
- Design patterns implementation (Singleton, Factory, Observer, etc.)
|
||||
- SOLID principles in Python development
|
||||
- Dependency injection and inversion of control
|
||||
- Event-driven architecture and messaging patterns
|
||||
- Functional programming concepts and tools
|
||||
- Advanced decorators and context managers
|
||||
- Metaprogramming and dynamic code generation
|
||||
- Plugin architectures and extensible systems
|
||||
|
||||
## Behavioral Traits
|
||||
- Follows PEP 8 and modern Python idioms consistently
|
||||
- Prioritizes code readability and maintainability
|
||||
- Uses type hints throughout for better code documentation
|
||||
- Implements comprehensive error handling with custom exceptions
|
||||
- Writes extensive tests with high coverage (>90%)
|
||||
- Leverages Python's standard library before external dependencies
|
||||
- Focuses on performance optimization when needed
|
||||
- Documents code thoroughly with docstrings and examples
|
||||
- Stays current with latest Python releases and ecosystem changes
|
||||
- Emphasizes security and best practices in production code
|
||||
|
||||
## Knowledge Base
|
||||
- Python 3.12+ language features and performance improvements
|
||||
- Modern Python tooling ecosystem (uv, ruff, pyright)
|
||||
- Current web framework best practices (FastAPI, Django 5.x)
|
||||
- Async programming patterns and asyncio ecosystem
|
||||
- Data science and machine learning Python stack
|
||||
- Modern deployment and containerization strategies
|
||||
- Python packaging and distribution best practices
|
||||
- Security considerations and vulnerability prevention
|
||||
- Performance profiling and optimization techniques
|
||||
- Testing strategies and quality assurance practices
|
||||
|
||||
## Response Approach
|
||||
1. **Analyze requirements** for modern Python best practices
|
||||
2. **Suggest current tools and patterns** from the 2024/2025 ecosystem
|
||||
3. **Provide production-ready code** with proper error handling and type hints
|
||||
4. **Include comprehensive tests** with pytest and appropriate fixtures
|
||||
5. **Consider performance implications** and suggest optimizations
|
||||
6. **Document security considerations** and best practices
|
||||
7. **Recommend modern tooling** for development workflow
|
||||
8. **Include deployment strategies** when applicable
|
||||
|
||||
## Example Interactions
|
||||
- "Help me migrate from pip to uv for package management"
|
||||
- "Optimize this Python code for better async performance"
|
||||
- "Design a FastAPI application with proper error handling and validation"
|
||||
- "Set up a modern Python project with ruff, mypy, and pytest"
|
||||
- "Implement a high-performance data processing pipeline"
|
||||
- "Create a production-ready Dockerfile for a Python application"
|
||||
- "Design a scalable background task system with Celery"
|
||||
- "Implement modern authentication patterns in FastAPI"
|
||||
@@ -0,0 +1,40 @@
|
||||
---
|
||||
name: python-testing-patterns
|
||||
description: "Implement comprehensive testing strategies with pytest, fixtures, mocking, and test-driven development. Use when writing Python tests, setting up test suites, or implementing testing best practices."
|
||||
risk: unknown
|
||||
source: community
|
||||
date_added: "2026-02-27"
|
||||
---
|
||||
|
||||
# Python Testing Patterns
|
||||
|
||||
Comprehensive guide to implementing robust testing strategies in Python using pytest, fixtures, mocking, parameterization, and test-driven development practices.
|
||||
|
||||
## Use this skill when
|
||||
|
||||
- Writing unit tests for Python code
|
||||
- Setting up test suites and test infrastructure
|
||||
- Implementing test-driven development (TDD)
|
||||
- Creating integration tests for APIs and services
|
||||
- Mocking external dependencies and services
|
||||
- Testing async code and concurrent operations
|
||||
- Setting up continuous testing in CI/CD
|
||||
- Implementing property-based testing
|
||||
- Testing database operations
|
||||
- Debugging failing tests
|
||||
|
||||
## Do not use this skill when
|
||||
|
||||
- The task is unrelated to python testing patterns
|
||||
- You need a different domain or tool outside this scope
|
||||
|
||||
## Instructions
|
||||
|
||||
- Clarify goals, constraints, and required inputs.
|
||||
- Apply relevant best practices and validate outcomes.
|
||||
- Provide actionable steps and verification.
|
||||
- If detailed examples are required, open `resources/implementation-playbook.md`.
|
||||
|
||||
## Resources
|
||||
|
||||
- `resources/implementation-playbook.md` for detailed patterns and examples.
|
||||
@@ -0,0 +1,906 @@
|
||||
# Python Testing Patterns Implementation Playbook
|
||||
|
||||
This file contains detailed patterns, checklists, and code samples referenced by the skill.
|
||||
|
||||
# Python Testing Patterns
|
||||
|
||||
Comprehensive guide to implementing robust testing strategies in Python using pytest, fixtures, mocking, parameterization, and test-driven development practices.
|
||||
|
||||
## When to Use This Skill
|
||||
|
||||
- Writing unit tests for Python code
|
||||
- Setting up test suites and test infrastructure
|
||||
- Implementing test-driven development (TDD)
|
||||
- Creating integration tests for APIs and services
|
||||
- Mocking external dependencies and services
|
||||
- Testing async code and concurrent operations
|
||||
- Setting up continuous testing in CI/CD
|
||||
- Implementing property-based testing
|
||||
- Testing database operations
|
||||
- Debugging failing tests
|
||||
|
||||
## Core Concepts
|
||||
|
||||
### 1. Test Types
|
||||
- **Unit Tests**: Test individual functions/classes in isolation
|
||||
- **Integration Tests**: Test interaction between components
|
||||
- **Functional Tests**: Test complete features end-to-end
|
||||
- **Performance Tests**: Measure speed and resource usage
|
||||
|
||||
### 2. Test Structure (AAA Pattern)
|
||||
- **Arrange**: Set up test data and preconditions
|
||||
- **Act**: Execute the code under test
|
||||
- **Assert**: Verify the results
|
||||
|
||||
### 3. Test Coverage
|
||||
- Measure what code is exercised by tests
|
||||
- Identify untested code paths
|
||||
- Aim for meaningful coverage, not just high percentages
|
||||
|
||||
### 4. Test Isolation
|
||||
- Tests should be independent
|
||||
- No shared state between tests
|
||||
- Each test should clean up after itself
|
||||
|
||||
## Quick Start
|
||||
|
||||
```python
|
||||
# test_example.py
|
||||
def add(a, b):
|
||||
return a + b
|
||||
|
||||
def test_add():
|
||||
"""Basic test example."""
|
||||
result = add(2, 3)
|
||||
assert result == 5
|
||||
|
||||
def test_add_negative():
|
||||
"""Test with negative numbers."""
|
||||
assert add(-1, 1) == 0
|
||||
|
||||
# Run with: pytest test_example.py
|
||||
```
|
||||
|
||||
## Fundamental Patterns
|
||||
|
||||
### Pattern 1: Basic pytest Tests
|
||||
|
||||
```python
|
||||
# test_calculator.py
|
||||
import pytest
|
||||
|
||||
class Calculator:
|
||||
"""Simple calculator for testing."""
|
||||
|
||||
def add(self, a: float, b: float) -> float:
|
||||
return a + b
|
||||
|
||||
def subtract(self, a: float, b: float) -> float:
|
||||
return a - b
|
||||
|
||||
def multiply(self, a: float, b: float) -> float:
|
||||
return a * b
|
||||
|
||||
def divide(self, a: float, b: float) -> float:
|
||||
if b == 0:
|
||||
raise ValueError("Cannot divide by zero")
|
||||
return a / b
|
||||
|
||||
|
||||
def test_addition():
|
||||
"""Test addition."""
|
||||
calc = Calculator()
|
||||
assert calc.add(2, 3) == 5
|
||||
assert calc.add(-1, 1) == 0
|
||||
assert calc.add(0, 0) == 0
|
||||
|
||||
|
||||
def test_subtraction():
|
||||
"""Test subtraction."""
|
||||
calc = Calculator()
|
||||
assert calc.subtract(5, 3) == 2
|
||||
assert calc.subtract(0, 5) == -5
|
||||
|
||||
|
||||
def test_multiplication():
|
||||
"""Test multiplication."""
|
||||
calc = Calculator()
|
||||
assert calc.multiply(3, 4) == 12
|
||||
assert calc.multiply(0, 5) == 0
|
||||
|
||||
|
||||
def test_division():
|
||||
"""Test division."""
|
||||
calc = Calculator()
|
||||
assert calc.divide(6, 3) == 2
|
||||
assert calc.divide(5, 2) == 2.5
|
||||
|
||||
|
||||
def test_division_by_zero():
|
||||
"""Test division by zero raises error."""
|
||||
calc = Calculator()
|
||||
with pytest.raises(ValueError, match="Cannot divide by zero"):
|
||||
calc.divide(5, 0)
|
||||
```
|
||||
|
||||
### Pattern 2: Fixtures for Setup and Teardown
|
||||
|
||||
```python
|
||||
# test_database.py
|
||||
import pytest
|
||||
from typing import Generator
|
||||
|
||||
class Database:
|
||||
"""Simple database class."""
|
||||
|
||||
def __init__(self, connection_string: str):
|
||||
self.connection_string = connection_string
|
||||
self.connected = False
|
||||
|
||||
def connect(self):
|
||||
"""Connect to database."""
|
||||
self.connected = True
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from database."""
|
||||
self.connected = False
|
||||
|
||||
def query(self, sql: str) -> list:
|
||||
"""Execute query."""
|
||||
if not self.connected:
|
||||
raise RuntimeError("Not connected")
|
||||
return [{"id": 1, "name": "Test"}]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db() -> Generator[Database, None, None]:
|
||||
"""Fixture that provides connected database."""
|
||||
# Setup
|
||||
database = Database("sqlite:///:memory:")
|
||||
database.connect()
|
||||
|
||||
# Provide to test
|
||||
yield database
|
||||
|
||||
# Teardown
|
||||
database.disconnect()
|
||||
|
||||
|
||||
def test_database_query(db):
|
||||
"""Test database query with fixture."""
|
||||
results = db.query("SELECT * FROM users")
|
||||
assert len(results) == 1
|
||||
assert results[0]["name"] == "Test"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def app_config():
|
||||
"""Session-scoped fixture - created once per test session."""
|
||||
return {
|
||||
"database_url": "postgresql://localhost/test",
|
||||
"api_key": "test-key",
|
||||
"debug": True
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def api_client(app_config):
|
||||
"""Module-scoped fixture - created once per test module."""
|
||||
# Setup expensive resource
|
||||
client = {"config": app_config, "session": "active"}
|
||||
yield client
|
||||
# Cleanup
|
||||
client["session"] = "closed"
|
||||
|
||||
|
||||
def test_api_client(api_client):
|
||||
"""Test using api client fixture."""
|
||||
assert api_client["session"] == "active"
|
||||
assert api_client["config"]["debug"] is True
|
||||
```
|
||||
|
||||
### Pattern 3: Parameterized Tests
|
||||
|
||||
```python
|
||||
# test_validation.py
|
||||
import pytest
|
||||
|
||||
def is_valid_email(email: str) -> bool:
|
||||
"""Check if email is valid."""
|
||||
return "@" in email and "." in email.split("@")[1]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("email,expected", [
|
||||
("user@example.com", True),
|
||||
("test.user@domain.co.uk", True),
|
||||
("invalid.email", False),
|
||||
("@example.com", False),
|
||||
("user@domain", False),
|
||||
("", False),
|
||||
])
|
||||
def test_email_validation(email, expected):
|
||||
"""Test email validation with various inputs."""
|
||||
assert is_valid_email(email) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("a,b,expected", [
|
||||
(2, 3, 5),
|
||||
(0, 0, 0),
|
||||
(-1, 1, 0),
|
||||
(100, 200, 300),
|
||||
(-5, -5, -10),
|
||||
])
|
||||
def test_addition_parameterized(a, b, expected):
|
||||
"""Test addition with multiple parameter sets."""
|
||||
from test_calculator import Calculator
|
||||
calc = Calculator()
|
||||
assert calc.add(a, b) == expected
|
||||
|
||||
|
||||
# Using pytest.param for special cases
|
||||
@pytest.mark.parametrize("value,expected", [
|
||||
pytest.param(1, True, id="positive"),
|
||||
pytest.param(0, False, id="zero"),
|
||||
pytest.param(-1, False, id="negative"),
|
||||
])
|
||||
def test_is_positive(value, expected):
|
||||
"""Test with custom test IDs."""
|
||||
assert (value > 0) == expected
|
||||
```
|
||||
|
||||
### Pattern 4: Mocking with unittest.mock
|
||||
|
||||
```python
|
||||
# test_api_client.py
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
import requests
|
||||
|
||||
class APIClient:
|
||||
"""Simple API client."""
|
||||
|
||||
def __init__(self, base_url: str):
|
||||
self.base_url = base_url
|
||||
|
||||
def get_user(self, user_id: int) -> dict:
|
||||
"""Fetch user from API."""
|
||||
response = requests.get(f"{self.base_url}/users/{user_id}")
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def create_user(self, data: dict) -> dict:
|
||||
"""Create new user."""
|
||||
response = requests.post(f"{self.base_url}/users", json=data)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
|
||||
def test_get_user_success():
|
||||
"""Test successful API call with mock."""
|
||||
client = APIClient("https://api.example.com")
|
||||
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {"id": 1, "name": "John Doe"}
|
||||
mock_response.raise_for_status.return_value = None
|
||||
|
||||
with patch("requests.get", return_value=mock_response) as mock_get:
|
||||
user = client.get_user(1)
|
||||
|
||||
assert user["id"] == 1
|
||||
assert user["name"] == "John Doe"
|
||||
mock_get.assert_called_once_with("https://api.example.com/users/1")
|
||||
|
||||
|
||||
def test_get_user_not_found():
|
||||
"""Test API call with 404 error."""
|
||||
client = APIClient("https://api.example.com")
|
||||
|
||||
mock_response = Mock()
|
||||
mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found")
|
||||
|
||||
with patch("requests.get", return_value=mock_response):
|
||||
with pytest.raises(requests.HTTPError):
|
||||
client.get_user(999)
|
||||
|
||||
|
||||
@patch("requests.post")
|
||||
def test_create_user(mock_post):
|
||||
"""Test user creation with decorator syntax."""
|
||||
client = APIClient("https://api.example.com")
|
||||
|
||||
mock_post.return_value.json.return_value = {"id": 2, "name": "Jane Doe"}
|
||||
mock_post.return_value.raise_for_status.return_value = None
|
||||
|
||||
user_data = {"name": "Jane Doe", "email": "jane@example.com"}
|
||||
result = client.create_user(user_data)
|
||||
|
||||
assert result["id"] == 2
|
||||
mock_post.assert_called_once()
|
||||
call_args = mock_post.call_args
|
||||
assert call_args.kwargs["json"] == user_data
|
||||
```
|
||||
|
||||
### Pattern 5: Testing Exceptions
|
||||
|
||||
```python
|
||||
# test_exceptions.py
|
||||
import pytest
|
||||
|
||||
def divide(a: float, b: float) -> float:
|
||||
"""Divide a by b."""
|
||||
if b == 0:
|
||||
raise ZeroDivisionError("Division by zero")
|
||||
if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
|
||||
raise TypeError("Arguments must be numbers")
|
||||
return a / b
|
||||
|
||||
|
||||
def test_zero_division():
|
||||
"""Test exception is raised for division by zero."""
|
||||
with pytest.raises(ZeroDivisionError):
|
||||
divide(10, 0)
|
||||
|
||||
|
||||
def test_zero_division_with_message():
|
||||
"""Test exception message."""
|
||||
with pytest.raises(ZeroDivisionError, match="Division by zero"):
|
||||
divide(5, 0)
|
||||
|
||||
|
||||
def test_type_error():
|
||||
"""Test type error exception."""
|
||||
with pytest.raises(TypeError, match="must be numbers"):
|
||||
divide("10", 5)
|
||||
|
||||
|
||||
def test_exception_info():
|
||||
"""Test accessing exception info."""
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
int("not a number")
|
||||
|
||||
assert "invalid literal" in str(exc_info.value)
|
||||
```
|
||||
|
||||
## Advanced Patterns
|
||||
|
||||
### Pattern 6: Testing Async Code
|
||||
|
||||
```python
|
||||
# test_async.py
|
||||
import pytest
|
||||
import asyncio
|
||||
|
||||
async def fetch_data(url: str) -> dict:
|
||||
"""Fetch data asynchronously."""
|
||||
await asyncio.sleep(0.1)
|
||||
return {"url": url, "data": "result"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_data():
|
||||
"""Test async function."""
|
||||
result = await fetch_data("https://api.example.com")
|
||||
assert result["url"] == "https://api.example.com"
|
||||
assert "data" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_fetches():
|
||||
"""Test concurrent async operations."""
|
||||
urls = ["url1", "url2", "url3"]
|
||||
tasks = [fetch_data(url) for url in urls]
|
||||
results = await asyncio.gather(*tasks)
|
||||
|
||||
assert len(results) == 3
|
||||
assert all("data" in r for r in results)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def async_client():
|
||||
"""Async fixture."""
|
||||
client = {"connected": True}
|
||||
yield client
|
||||
client["connected"] = False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_with_async_fixture(async_client):
|
||||
"""Test using async fixture."""
|
||||
assert async_client["connected"] is True
|
||||
```
|
||||
|
||||
### Pattern 7: Monkeypatch for Testing
|
||||
|
||||
```python
|
||||
# test_environment.py
|
||||
import os
|
||||
import pytest
|
||||
|
||||
def get_database_url() -> str:
|
||||
"""Get database URL from environment."""
|
||||
return os.environ.get("DATABASE_URL", "sqlite:///:memory:")
|
||||
|
||||
|
||||
def test_database_url_default():
|
||||
"""Test default database URL."""
|
||||
# Will use actual environment variable if set
|
||||
url = get_database_url()
|
||||
assert url
|
||||
|
||||
|
||||
def test_database_url_custom(monkeypatch):
|
||||
"""Test custom database URL with monkeypatch."""
|
||||
monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test")
|
||||
assert get_database_url() == "postgresql://localhost/test"
|
||||
|
||||
|
||||
def test_database_url_not_set(monkeypatch):
|
||||
"""Test when env var is not set."""
|
||||
monkeypatch.delenv("DATABASE_URL", raising=False)
|
||||
assert get_database_url() == "sqlite:///:memory:"
|
||||
|
||||
|
||||
class Config:
|
||||
"""Configuration class."""
|
||||
|
||||
def __init__(self):
|
||||
self.api_key = "production-key"
|
||||
|
||||
def get_api_key(self):
|
||||
return self.api_key
|
||||
|
||||
|
||||
def test_monkeypatch_attribute(monkeypatch):
|
||||
"""Test monkeypatching object attributes."""
|
||||
config = Config()
|
||||
monkeypatch.setattr(config, "api_key", "test-key")
|
||||
assert config.get_api_key() == "test-key"
|
||||
```
|
||||
|
||||
### Pattern 8: Temporary Files and Directories
|
||||
|
||||
```python
|
||||
# test_file_operations.py
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
def save_data(filepath: Path, data: str):
|
||||
"""Save data to file."""
|
||||
filepath.write_text(data)
|
||||
|
||||
|
||||
def load_data(filepath: Path) -> str:
|
||||
"""Load data from file."""
|
||||
return filepath.read_text()
|
||||
|
||||
|
||||
def test_file_operations(tmp_path):
|
||||
"""Test file operations with temporary directory."""
|
||||
# tmp_path is a pathlib.Path object
|
||||
test_file = tmp_path / "test_data.txt"
|
||||
|
||||
# Save data
|
||||
save_data(test_file, "Hello, World!")
|
||||
|
||||
# Verify file exists
|
||||
assert test_file.exists()
|
||||
|
||||
# Load and verify data
|
||||
data = load_data(test_file)
|
||||
assert data == "Hello, World!"
|
||||
|
||||
|
||||
def test_multiple_files(tmp_path):
|
||||
"""Test with multiple temporary files."""
|
||||
files = {
|
||||
"file1.txt": "Content 1",
|
||||
"file2.txt": "Content 2",
|
||||
"file3.txt": "Content 3"
|
||||
}
|
||||
|
||||
for filename, content in files.items():
|
||||
filepath = tmp_path / filename
|
||||
save_data(filepath, content)
|
||||
|
||||
# Verify all files created
|
||||
assert len(list(tmp_path.iterdir())) == 3
|
||||
|
||||
# Verify contents
|
||||
for filename, expected_content in files.items():
|
||||
filepath = tmp_path / filename
|
||||
assert load_data(filepath) == expected_content
|
||||
```
|
||||
|
||||
### Pattern 9: Custom Fixtures and Conftest
|
||||
|
||||
```python
|
||||
# conftest.py
|
||||
"""Shared fixtures for all tests."""
|
||||
import pytest
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def database_url():
|
||||
"""Provide database URL for all tests."""
|
||||
return "postgresql://localhost/test_db"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_database(database_url):
|
||||
"""Auto-use fixture that runs before each test."""
|
||||
# Setup: Clear database
|
||||
print(f"Clearing database: {database_url}")
|
||||
yield
|
||||
# Teardown: Clean up
|
||||
print("Test completed")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_user():
|
||||
"""Provide sample user data."""
|
||||
return {
|
||||
"id": 1,
|
||||
"name": "Test User",
|
||||
"email": "test@example.com"
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_users():
|
||||
"""Provide list of sample users."""
|
||||
return [
|
||||
{"id": 1, "name": "User 1"},
|
||||
{"id": 2, "name": "User 2"},
|
||||
{"id": 3, "name": "User 3"},
|
||||
]
|
||||
|
||||
|
||||
# Parametrized fixture
|
||||
@pytest.fixture(params=["sqlite", "postgresql", "mysql"])
|
||||
def db_backend(request):
|
||||
"""Fixture that runs tests with different database backends."""
|
||||
return request.param
|
||||
|
||||
|
||||
def test_with_db_backend(db_backend):
|
||||
"""This test will run 3 times with different backends."""
|
||||
print(f"Testing with {db_backend}")
|
||||
assert db_backend in ["sqlite", "postgresql", "mysql"]
|
||||
```
|
||||
|
||||
### Pattern 10: Property-Based Testing
|
||||
|
||||
```python
|
||||
# test_properties.py
|
||||
from hypothesis import given, strategies as st
|
||||
import pytest
|
||||
|
||||
def reverse_string(s: str) -> str:
|
||||
"""Reverse a string."""
|
||||
return s[::-1]
|
||||
|
||||
|
||||
@given(st.text())
|
||||
def test_reverse_twice_is_original(s):
|
||||
"""Property: reversing twice returns original."""
|
||||
assert reverse_string(reverse_string(s)) == s
|
||||
|
||||
|
||||
@given(st.text())
|
||||
def test_reverse_length(s):
|
||||
"""Property: reversed string has same length."""
|
||||
assert len(reverse_string(s)) == len(s)
|
||||
|
||||
|
||||
@given(st.integers(), st.integers())
|
||||
def test_addition_commutative(a, b):
|
||||
"""Property: addition is commutative."""
|
||||
assert a + b == b + a
|
||||
|
||||
|
||||
@given(st.lists(st.integers()))
|
||||
def test_sorted_list_properties(lst):
|
||||
"""Property: sorted list is ordered."""
|
||||
sorted_lst = sorted(lst)
|
||||
|
||||
# Same length
|
||||
assert len(sorted_lst) == len(lst)
|
||||
|
||||
# All elements present
|
||||
assert set(sorted_lst) == set(lst)
|
||||
|
||||
# Is ordered
|
||||
for i in range(len(sorted_lst) - 1):
|
||||
assert sorted_lst[i] <= sorted_lst[i + 1]
|
||||
```
|
||||
|
||||
## Testing Best Practices
|
||||
|
||||
### Test Organization
|
||||
|
||||
```python
|
||||
# tests/
|
||||
# __init__.py
|
||||
# conftest.py # Shared fixtures
|
||||
# test_unit/ # Unit tests
|
||||
# test_models.py
|
||||
# test_utils.py
|
||||
# test_integration/ # Integration tests
|
||||
# test_api.py
|
||||
# test_database.py
|
||||
# test_e2e/ # End-to-end tests
|
||||
# test_workflows.py
|
||||
```
|
||||
|
||||
### Test Naming
|
||||
|
||||
```python
|
||||
# Good test names
|
||||
def test_user_creation_with_valid_data():
|
||||
"""Clear name describes what is being tested."""
|
||||
pass
|
||||
|
||||
|
||||
def test_login_fails_with_invalid_password():
|
||||
"""Name describes expected behavior."""
|
||||
pass
|
||||
|
||||
|
||||
def test_api_returns_404_for_missing_resource():
|
||||
"""Specific about inputs and expected outcomes."""
|
||||
pass
|
||||
|
||||
|
||||
# Bad test names
|
||||
def test_1(): # Not descriptive
|
||||
pass
|
||||
|
||||
|
||||
def test_user(): # Too vague
|
||||
pass
|
||||
|
||||
|
||||
def test_function(): # Doesn't explain what's tested
|
||||
pass
|
||||
```
|
||||
|
||||
### Test Markers
|
||||
|
||||
```python
|
||||
# test_markers.py
|
||||
import pytest
|
||||
|
||||
@pytest.mark.slow
|
||||
def test_slow_operation():
|
||||
"""Mark slow tests."""
|
||||
import time
|
||||
time.sleep(2)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_database_integration():
|
||||
"""Mark integration tests."""
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.skip(reason="Feature not implemented yet")
|
||||
def test_future_feature():
|
||||
"""Skip tests temporarily."""
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="Unix only test")
|
||||
def test_unix_specific():
|
||||
"""Conditional skip."""
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.xfail(reason="Known bug #123")
|
||||
def test_known_bug():
|
||||
"""Mark expected failures."""
|
||||
assert False
|
||||
|
||||
|
||||
# Run with:
|
||||
# pytest -m slow # Run only slow tests
|
||||
# pytest -m "not slow" # Skip slow tests
|
||||
# pytest -m integration # Run integration tests
|
||||
```
|
||||
|
||||
### Coverage Reporting
|
||||
|
||||
```bash
|
||||
# Install coverage
|
||||
pip install pytest-cov
|
||||
|
||||
# Run tests with coverage
|
||||
pytest --cov=myapp tests/
|
||||
|
||||
# Generate HTML report
|
||||
pytest --cov=myapp --cov-report=html tests/
|
||||
|
||||
# Fail if coverage below threshold
|
||||
pytest --cov=myapp --cov-fail-under=80 tests/
|
||||
|
||||
# Show missing lines
|
||||
pytest --cov=myapp --cov-report=term-missing tests/
|
||||
```
|
||||
|
||||
## Testing Database Code
|
||||
|
||||
```python
|
||||
# test_database_models.py
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, Column, Integer, String
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker, Session
|
||||
|
||||
Base = declarative_base()
|
||||
|
||||
|
||||
class User(Base):
|
||||
"""User model."""
|
||||
__tablename__ = "users"
|
||||
|
||||
id = Column(Integer, primary_key=True)
|
||||
name = Column(String(50))
|
||||
email = Column(String(100), unique=True)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def db_session() -> Session:
|
||||
"""Create in-memory database for testing."""
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
|
||||
SessionLocal = sessionmaker(bind=engine)
|
||||
session = SessionLocal()
|
||||
|
||||
yield session
|
||||
|
||||
session.close()
|
||||
|
||||
|
||||
def test_create_user(db_session):
|
||||
"""Test creating a user."""
|
||||
user = User(name="Test User", email="test@example.com")
|
||||
db_session.add(user)
|
||||
db_session.commit()
|
||||
|
||||
assert user.id is not None
|
||||
assert user.name == "Test User"
|
||||
|
||||
|
||||
def test_query_user(db_session):
|
||||
"""Test querying users."""
|
||||
user1 = User(name="User 1", email="user1@example.com")
|
||||
user2 = User(name="User 2", email="user2@example.com")
|
||||
|
||||
db_session.add_all([user1, user2])
|
||||
db_session.commit()
|
||||
|
||||
users = db_session.query(User).all()
|
||||
assert len(users) == 2
|
||||
|
||||
|
||||
def test_unique_email_constraint(db_session):
|
||||
"""Test unique email constraint."""
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
user1 = User(name="User 1", email="same@example.com")
|
||||
user2 = User(name="User 2", email="same@example.com")
|
||||
|
||||
db_session.add(user1)
|
||||
db_session.commit()
|
||||
|
||||
db_session.add(user2)
|
||||
|
||||
with pytest.raises(IntegrityError):
|
||||
db_session.commit()
|
||||
```
|
||||
|
||||
## CI/CD Integration
|
||||
|
||||
```yaml
|
||||
# .github/workflows/test.yml
|
||||
name: Tests
|
||||
|
||||
on: [push, pull_request]
|
||||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
strategy:
|
||||
matrix:
|
||||
python-version: ["3.9", "3.10", "3.11", "3.12"]
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v3
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@v4
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
|
||||
- name: Install dependencies
|
||||
run: |
|
||||
pip install -e ".[dev]"
|
||||
pip install pytest pytest-cov
|
||||
|
||||
- name: Run tests
|
||||
run: |
|
||||
pytest --cov=myapp --cov-report=xml
|
||||
|
||||
- name: Upload coverage
|
||||
uses: codecov/codecov-action@v3
|
||||
with:
|
||||
file: ./coverage.xml
|
||||
```
|
||||
|
||||
## Configuration Files
|
||||
|
||||
```ini
|
||||
# pytest.ini
|
||||
[pytest]
|
||||
testpaths = tests
|
||||
python_files = test_*.py
|
||||
python_classes = Test*
|
||||
python_functions = test_*
|
||||
addopts =
|
||||
-v
|
||||
--strict-markers
|
||||
--tb=short
|
||||
--cov=myapp
|
||||
--cov-report=term-missing
|
||||
markers =
|
||||
slow: marks tests as slow
|
||||
integration: marks integration tests
|
||||
unit: marks unit tests
|
||||
e2e: marks end-to-end tests
|
||||
```
|
||||
|
||||
```toml
|
||||
# pyproject.toml
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
python_files = ["test_*.py"]
|
||||
addopts = [
|
||||
"-v",
|
||||
"--cov=myapp",
|
||||
"--cov-report=term-missing",
|
||||
]
|
||||
|
||||
[tool.coverage.run]
|
||||
source = ["myapp"]
|
||||
omit = ["*/tests/*", "*/migrations/*"]
|
||||
|
||||
[tool.coverage.report]
|
||||
exclude_lines = [
|
||||
"pragma: no cover",
|
||||
"def __repr__",
|
||||
"raise AssertionError",
|
||||
"raise NotImplementedError",
|
||||
]
|
||||
```
|
||||
|
||||
## Resources
|
||||
|
||||
- **pytest documentation**: https://docs.pytest.org/
|
||||
- **unittest.mock**: https://docs.python.org/3/library/unittest.mock.html
|
||||
- **hypothesis**: Property-based testing
|
||||
- **pytest-asyncio**: Testing async code
|
||||
- **pytest-cov**: Coverage reporting
|
||||
- **pytest-mock**: pytest wrapper for mock
|
||||
|
||||
## Best Practices Summary
|
||||
|
||||
1. **Write tests first** (TDD) or alongside code
|
||||
2. **One assertion per test** when possible
|
||||
3. **Use descriptive test names** that explain behavior
|
||||
4. **Keep tests independent** and isolated
|
||||
5. **Use fixtures** for setup and teardown
|
||||
6. **Mock external dependencies** appropriately
|
||||
7. **Parametrize tests** to reduce duplication
|
||||
8. **Test edge cases** and error conditions
|
||||
9. **Measure coverage** but focus on quality
|
||||
10. **Run tests in CI/CD** on every commit
|
||||
Reference in New Issue
Block a user