This document outlines the coding conventions and best practices for Python applications in our organization, with special consideration for AI-assisted development. Following these guidelines will ensure consistency, maintainability, and optimal collaboration between human developers and AI assistants.
- Project Structure
- Type Hints
- API Development with FastAPI
- HTTP Client: HTTPX
- Web Interface Development
- Environment Variables Management
- Logging
- Scalability Considerations
- Testing
- Documentation
project-name/
├── alembic/ # Database migrations
├── src/
│ ├── domain1/
│ │ ├── router.py # FastAPI router
│ │ ├── schemas.py # Pydantic models
│ │ ├── models.py # DB models
│ │ ├── dependencies.py # Router dependencies
│ │ ├── config.py # Domain-specific configs
│ │ ├── constants.py # Domain-specific constants
│ │ ├── exceptions.py # Custom exceptions
│ │ ├── service.py # Business logic
│ │ └── utils.py # Helper functions
│ ├── domain2/
│ │ └── ...
│ ├── config.py # Global configs
│ ├── models.py # Shared models
│ ├── exceptions.py # Global exceptions
│ ├── logging.py # Logging configuration
│ ├── database.py # DB connection setup
│ └── main.py # Application entry point
├── tests/
│ ├── domain1/
│ ├── domain2/
│ └── conftest.py
├── templates/ # HTML templates
│ └── ...
├── static/
│ ├── css/
│ └── js/
├── tailwindcss/ # TailwindCSS configuration
├── .env # Environment variables (not in git)
├── .env.example # Example environment variables
├── requirements.txt # Production dependencies
├── requirements-dev.txt # Development dependencies
├── Dockerfile # Container definition
└── README.md
When importing from other packages, use explicit module names:
from src.auth import constants as auth_constants
from src.notifications import service as notification_serviceAlways use type hints to improve code clarity, enable better IDE support, and facilitate AI code understanding and generation.
# Variables
age: int = 1
names: list[str] = ["Alice", "Bob"]
user_data: dict[str, Any] = {"name": "Alice", "age": 30}
# Functions
def calculate_area(length: float, width: float) -> float:
return length * width- Use
TypeAliasfor type aliases:
from typing import TypeAlias
IntList: TypeAlias = list[int]- Use the appropriate collection type hints:
# For Python 3.9+
values: list[int] = [1, 2, 3]
mappings: dict[str, float] = {"field": 2.0}
fixed_tuple: tuple[int, str, float] = (3, "yes", 7.5)
variable_tuple: tuple[int, ...] = (1, 2, 3)- Use
Anywhen a type cannot be expressed appropriately with the current type system:
from typing import Any
def process_unknown_data(data: Any) -> str:
return str(data)- Use
objectinstead ofAnywhen a function accepts any possible object but doesn't need specific operations:
def log_value(value: object) -> None:
print(f"Value: {value}")- Prefer protocols and abstract types for arguments, concrete types for return values:
from typing import Sequence, Iterable, Mapping
def process_items(items: Sequence[int]) -> list[str]:
return [str(item) for item in items]- Use mypy for static type checking:
python -m pip install mypy
mypy src/from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI(title="My API", version="1.0.0")
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Adjust in production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)Properly handle async operations to prevent blocking:
# Good - Non-blocking async route
@app.get("/perfect-endpoint")
async def perfect_endpoint():
await asyncio.sleep(1) # Non-blocking I/O operation
return {"status": "success"}
# Good - Sync route for blocking operations
@app.get("/good-endpoint")
def good_endpoint():
time.sleep(1) # Blocking operation, but FastAPI runs it in a thread
return {"status": "success"}
# BAD - Blocking operation in async route
@app.get("/bad-endpoint")
async def bad_endpoint():
time.sleep(1) # This blocks the event loop!
return {"status": "success"}from fastapi import HTTPException
from pydantic import BaseModel
class ErrorResponse(BaseModel):
detail: str
code: str
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.detail, "code": "HTTP_ERROR"}
)Use Uvicorn with multiple workers for production:
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4For CPU-bound applications, match workers to the number of CPU cores. For I/O-bound applications, you can use more workers than cores.
Always prefer HTTPX over requests for making HTTP requests, especially for modern Python applications.
import httpx
def fetch_data(url: str) -> dict:
with httpx.Client(timeout=10.0) as client:
response = client.get(url)
response.raise_for_status()
return response.json()import httpx
import asyncio
async def fetch_data_async(url: str) -> dict:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(url)
response.raise_for_status()
return response.json()- Use a single client instance for connection pooling:
# Global client for reuse
http_client = httpx.Client()
# Close on application shutdown
@app.on_event("shutdown")
def shutdown_event():
http_client.close()- Always set appropriate timeouts:
client = httpx.Client(
timeout=httpx.Timeout(5.0, connect=3.0)
)- Use structured error handling:
try:
response = client.get(url)
response.raise_for_status()
except httpx.RequestError as exc:
logger.error(f"Request failed: {exc}")
except httpx.HTTPStatusError as exc:
logger.error(f"HTTP error: {exc}")- Install TailwindCSS:
npm install tailwindcss- Create a tailwind configuration file:
// tailwind.config.js
module.exports = {
content: ["../templates/**/*.html"],
theme: {
extend: {},
},
plugins: [],
}- Set up the FastAPI template system:
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
@app.get("/")
async def index(request: Request):
return templates.TemplateResponse("base.html", {"request": request})- Create HTML templates with TailwindCSS classes:
<!-- templates/base.html -->
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>My App</title>
<link href="/static/css/tailwind.css" rel="stylesheet">
</head>
<body class="bg-gray-100">
<div class="container mx-auto p-4">
<h1 class="text-2xl font-bold text-blue-600">Hello, FastAPI with Tailwind!</h1>
</div>
</body>
</html>Use python-dotenv for environment variable management.
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Access environment variables
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
DEBUG = os.getenv("DEBUG", "False").lower() == "true"- Keep
.envout of version control:
# .gitignore
.env
- Provide
.env.examplewith dummy values:
# .env.example
DATABASE_URL=postgresql://user:password@localhost/dbname
DEBUG=False
SECRET_KEY=replace_with_secure_key
- Validate required environment variables on startup:
def validate_env_vars():
required_vars = ["SECRET_KEY", "DATABASE_URL"]
missing = [var for var in required_vars if not os.getenv(var)]
if missing:
raise RuntimeError(f"Missing required environment variables: {', '.join(missing)}")import logging
from logging.config import dictConfig
LOGGING_CONFIG = {
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"default": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
},
"json": {
"format": '{"timestamp": "%(asctime)s", "logger": "%(name)s", "level": "%(levelname)s", "message": "%(message)s"}'
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"formatter": "default"
},
"file": {
"class": "logging.handlers.RotatingFileHandler",
"filename": "app.log",
"maxBytes": 10485760, # 10MB
"backupCount": 5,
"formatter": "json"
},
},
"loggers": {
"app": {
"handlers": ["console", "file"],
"level": "INFO",
"propagate": False
},
},
}
dictConfig(LOGGING_CONFIG)
logger = logging.getLogger("app")@app.middleware("http")
async def log_requests(request: Request, call_next):
import uuid
request_id = str(uuid.uuid4())
logger.info(f"Request {request_id}: {request.method} {request.url}")
response = await call_next(request)
logger.info(f"Response {request_id}: {response.status_code}")
return response- Use appropriate log levels:
logger.debug("Detailed information for debugging")
logger.info("Confirmation of expected events")
logger.warning("Something unexpected but the application still works")
logger.error("An error that prevents a function from working")
logger.critical("An error that prevents the application from working")- Mask sensitive information:
def mask_email(email: str) -> str:
username, domain = email.split('@')
return f"{username[:2]}{'*' * (len(username) - 2)}@{domain}"
logger.info(f"Processing request for user: {mask_email(user.email)}")- Log structured data for easier parsing:
import json
def log_event(event_type: str, data: dict) -> None:
logger.info(f"{event_type}: {json.dumps(data)}")- Avoid storing session state in the application memory:
# BAD - In-memory state
user_sessions = {}
# GOOD - Use external session store
from fastapi_sessions.backends.redis import RedisBackend- Use external storage for shared state:
import redis
redis_client = redis.Redis.from_url(os.getenv("REDIS_URL"))
def increment_counter(key: str) -> int:
return redis_client.incr(key)- Design for horizontal scaling:
# Configure connection pooling appropriately
from sqlalchemy.pool import QueuePool
engine = create_engine(
DATABASE_URL,
poolclass=QueuePool,
pool_size=5,
max_overflow=10
)- Configure Uvicorn with multiple workers:
# In a deployment script
import multiprocessing
workers = multiprocessing.cpu_count() * 2 + 1
# For Gunicorn with Uvicorn workers
# gunicorn -w {workers} -k uvicorn.workers.UvicornWorker main:app- Ensure workers don't interfere with each other:
# Use atomic operations for shared resources
async def increment_counter(key: str) -> int:
# Using Redis INCR which is atomic
return await redis.incr(key)tests/
├── conftest.py # Shared fixtures
├── test_main.py # Application-level tests
└── domain1/
├── test_api.py # API tests
├── test_models.py # Model tests
└── test_services.py # Service tests
import pytest
from httpx import AsyncClient
from typing import AsyncGenerator
@pytest.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
from src.main import app
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.mark.asyncio
async def test_read_main(client: AsyncClient) -> None:
response = await client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}Use Google-style docstrings for better AI understanding:
def calculate_area(length: float, width: float) -> float:
"""Calculate the area of a rectangle.
Args:
length: The length of the rectangle.
width: The width of the rectangle.
Returns:
The area of the rectangle.
Raises:
ValueError: If length or width is negative.
"""
if length < 0 or width < 0:
raise ValueError("Length and width must be positive")
return length * widthEnsure your README.md includes:
- Project description and purpose
- Installation instructions
- Usage examples
- Configuration options
- Development setup
- Testing instructions
- Contribution guidelines
- License information
By following these conventions, you'll create Python applications that are maintainable, scalable, and optimally suited for AI-assisted development.