Instrucciones de Copilot para Proyectos Python/FastAPI¶
Requisitos de Proceso Importantes¶
- SIEMPRE presenta un plan detallado y espera aprobación explícita antes de implementar cualquier cambio de código
- No proceder con la implementación hasta recibir confirmación del usuario
- Al presentar el plan, proporciona un desglose paso a paso de todos los archivos a crear o modificar
- Pregunta directamente: "¿Apruebas este plan antes de proceder con la implementación?"
Convenciones de Nomenclatura¶
🔤 Patrones Generales de Nomenclatura¶
| Tipo de Elemento | Patrón/Estilo | Ejemplo(s) |
|---|---|---|
| Variables | snake_case |
user_id, total_amount |
| Funciones | snake_case |
get_user_data(), is_available() |
| Constantes | UPPER_SNAKE_CASE |
MAX_TIMEOUT, DEFAULT_PORT |
| Clases | PascalCase |
UserService, DatabaseConnection |
| Módulos/Archivos | snake_case |
user_service.py, database_config.py |
| Paquetes | lowercase |
models, services, utils |
| Variables Privadas | _snake_case |
_internal_state, _helper_method() |
| Variables Muy Privadas | __snake_case |
__private_attr |
🧠 Mejores Prácticas de Nomenclatura¶
- Usar nombres descriptivos y completos (evitar abreviaciones como
usr,cfg) - Para funciones booleanas, usar prefijos como
is_,has_,can_,should_ - Para funciones que retornan datos, usar prefijos como
get_,fetch_,load_ - Para funciones que modifican estado, usar verbos como
create_,update_,delete_ - Evitar nombres que coincidan con palabras reservadas de Python
- Usar nombres en inglés consistentemente en todo el proyecto
✅ Estilo de Código¶
📐 Formato y Estructura¶
- Usar 4 espacios para indentación (nunca tabs)
- Longitud máxima de línea: 88 caracteres (compatible con Black)
- Usar Black para formateo automático de código
- Usar isort para organización de imports
- Seguir PEP 8 como guía base de estilo
🔠 Codificación¶
- Usar codificación UTF-8 para todos los archivos
- Incluir
# -*- coding: utf-8 -*-si es necesario para compatibilidad
💡 Declaraciones y Imports¶
- Agrupar imports en este orden:
- Librerías estándar de Python
- Librerías de terceros
- Imports locales del proyecto
- Usar imports absolutos cuando sea posible
- Evitar imports con
*(wildcard imports) - Un import por línea para imports múltiples del mismo módulo
# Correcto
import os
import sys
from typing import List, Dict, Optional
import fastapi
import pydantic
from sqlalchemy import create_engine
from app.models import User
from app.services import UserService
🏗️ Estructura del Proyecto¶
Estructura Recomendada para FastAPI¶
project_root/
├── app/
│ ├── __init__.py
│ ├── main.py # Punto de entrada de FastAPI
│ ├── config/
│ │ ├── __init__.py
│ │ ├── settings.py # Configuración de la aplicación
│ │ └── database.py # Configuración de base de datos
│ ├── models/
│ │ ├── __init__.py
│ │ ├── base.py # Modelo base
│ │ └── user.py # Modelos específicos
│ ├── schemas/
│ │ ├── __init__.py
│ │ └── user.py # Esquemas Pydantic
│ ├── services/
│ │ ├── __init__.py
│ │ └── user_service.py # Lógica de negocio
│ ├── repositories/
│ │ ├── __init__.py
│ │ └── user_repository.py # Acceso a datos
│ ├── api/
│ │ ├── __init__.py
│ │ ├── dependencies.py # Dependencias de FastAPI
│ │ └── v1/
│ │ ├── __init__.py
│ │ ├── router.py # Router principal
│ │ └── endpoints/
│ │ ├── __init__.py
│ │ └── users.py # Endpoints específicos
│ ├── core/
│ │ ├── __init__.py
│ │ ├── security.py # Autenticación y autorización
│ │ └── exceptions.py # Excepciones personalizadas
│ └── utils/
│ ├── __init__.py
│ └── helpers.py # Funciones utilitarias
├── tests/
│ ├── __init__.py
│ ├── conftest.py # Configuración de pytest
│ ├── test_main.py
│ └── api/
│ └── test_users.py
├── requirements/
│ ├── base.txt # Dependencias base
│ ├── dev.txt # Dependencias de desarrollo
│ └── prod.txt # Dependencias de producción
├── .env.example # Variables de entorno ejemplo
├── .gitignore
├── pyproject.toml # Configuración del proyecto
├── README.md
└── Dockerfile
📝 Estándares de Documentación¶
Docstrings¶
- Usar Google Style o NumPy Style para docstrings
- Documentar todas las funciones públicas, clases y módulos
- Incluir tipos de parámetros y valores de retorno
def get_user_by_id(user_id: int, db: Session) -> Optional[User]:
"""
Retrieve a user by their ID.
Args:
user_id (int): The unique identifier for the user.
db (Session): Database session for the query.
Returns:
Optional[User]: The user object if found, None otherwise.
Raises:
DatabaseError: If there's an issue with the database connection.
"""
try:
return db.query(User).filter(User.id == user_id).first()
except Exception as e:
raise DatabaseError(f"Failed to retrieve user {user_id}: {str(e)}")
Type Hints¶
- Usar type hints en todas las funciones y métodos
- Importar tipos desde
typingcuando sea necesario - Usar
Optionalpara valores que pueden ser None - Usar
Unionpara múltiples tipos posibles
from typing import List, Dict, Optional, Union
from datetime import datetime
def process_user_data(
users: List[Dict[str, Union[str, int]]],
created_after: Optional[datetime] = None
) -> List[User]:
"""Process and filter user data."""
# Implementation here
pass
🧪 Guías de Testing¶
Estructura de Tests¶
- Usar pytest como framework de testing
- Organizar tests en estructura similar al código fuente
- Usar fixtures para setup y teardown
- Nombrar tests de forma descriptiva:
test_should_do_x_when_y
Patrones de Testing¶
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.config.database import get_db
from app.models.base import Base
# Test database setup
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture
def db_session():
"""Create a test database session."""
Base.metadata.create_all(bind=engine)
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
Base.metadata.drop_all(bind=engine)
@pytest.fixture
def client(db_session):
"""Create a test client."""
def override_get_db():
try:
yield db_session
finally:
db_session.close()
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as test_client:
yield test_client
def test_should_create_user_when_valid_data_provided(client, db_session):
"""Test user creation with valid data."""
# Arrange
user_data = {
"email": "test@example.com",
"name": "Test User"
}
# Act
response = client.post("/api/v1/users/", json=user_data)
# Assert
assert response.status_code == 201
assert response.json()["email"] == user_data["email"]
📦 Gestión de Dependencias¶
Uso de Requirements¶
- Separar dependencias por entorno (base, dev, prod)
- Fijar versiones específicas para producción
- Usar
pip-toolspara gestión avanzada de dependencias
# requirements/base.txt
fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
sqlalchemy==2.0.23
alembic==1.13.0
python-multipart==0.0.6
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
# requirements/dev.txt
-r base.txt
pytest==7.4.3
pytest-asyncio==0.21.1
black==23.11.0
isort==5.12.0
flake8==6.1.0
mypy==1.7.1
pre-commit==3.6.0
# requirements/prod.txt
-r base.txt
gunicorn==21.2.0
psycopg2-binary==2.9.9
Configuración con pyproject.toml¶
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "my-fastapi-app"
version = "0.1.0"
description = "FastAPI application with best practices"
authors = [{name = "Your Name", email = "your.email@example.com"}]
dependencies = [
"fastapi>=0.104.0",
"uvicorn[standard]>=0.24.0",
"pydantic>=2.5.0",
]
[tool.black]
line-length = 88
target-version = ['py311']
include = '\.pyi?$'
[tool.isort]
profile = "black"
multi_line_output = 3
line_length = 88
[tool.mypy]
python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
🌐 Documentación de API¶
FastAPI y OpenAPI¶
- Usar decoradores de FastAPI para documentación automática
- Incluir descripciones claras en endpoints
- Documentar todos los códigos de respuesta posibles
- Usar esquemas Pydantic para validación y documentación
from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel
from typing import List
app = FastAPI(
title="User Management API",
description="API for managing users in the system",
version="1.0.0"
)
class UserCreate(BaseModel):
"""Schema for creating a new user."""
email: str
name: str
age: Optional[int] = None
class UserResponse(BaseModel):
"""Schema for user response."""
id: int
email: str
name: str
age: Optional[int] = None
created_at: datetime
@app.post(
"/api/v1/users/",
response_model=UserResponse,
status_code=status.HTTP_201_CREATED,
summary="Create a new user",
description="Create a new user with the provided information",
responses={
201: {"description": "User created successfully"},
400: {"description": "Invalid input data"},
409: {"description": "User already exists"}
}
)
async def create_user(
user_data: UserCreate,
db: Session = Depends(get_db)
) -> UserResponse:
"""
Create a new user in the system.
- **email**: User's email address (must be unique)
- **name**: User's full name
- **age**: User's age (optional)
"""
# Implementation here
pass
🔒 Manejo de Errores y Seguridad¶
Excepciones Personalizadas¶
from fastapi import HTTPException, status
class BaseAPIException(Exception):
"""Base exception for API errors."""
def __init__(self, message: str, status_code: int = 500):
self.message = message
self.status_code = status_code
super().__init__(self.message)
class UserNotFoundError(BaseAPIException):
"""Raised when a user is not found."""
def __init__(self, user_id: int):
super().__init__(
message=f"User with ID {user_id} not found",
status_code=status.HTTP_404_NOT_FOUND
)
class ValidationError(BaseAPIException):
"""Raised when input validation fails."""
def __init__(self, message: str):
super().__init__(
message=f"Validation error: {message}",
status_code=status.HTTP_400_BAD_REQUEST
)
# Exception handler
@app.exception_handler(BaseAPIException)
async def api_exception_handler(request: Request, exc: BaseAPIException):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.message, "status_code": exc.status_code}
)
Configuración de Seguridad¶
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt
from datetime import datetime, timedelta
security = HTTPBearer()
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, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Get current authenticated user."""
try:
payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
user_id: int = payload.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials"
)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials"
)
# Get user from database
user = get_user_by_id(user_id)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found"
)
return user
🚀 Rendimiento y Optimización¶
Mejores Prácticas de Rendimiento¶
- Usar async/await para operaciones I/O
- Implementar connection pooling para bases de datos
- Usar caching para datos frecuentemente accedidos
- Implementar paginación para listas grandes
- Usar background tasks para operaciones pesadas
from fastapi import BackgroundTasks
from sqlalchemy.ext.asyncio import AsyncSession
import asyncio
from functools import lru_cache
# Async database operations
async def get_users_async(db: AsyncSession, skip: int = 0, limit: int = 100):
"""Get users asynchronously with pagination."""
result = await db.execute(
select(User).offset(skip).limit(limit)
)
return result.scalars().all()
# Caching
@lru_cache(maxsize=128)
def get_expensive_computation(param: str) -> str:
"""Cached expensive computation."""
# Expensive operation here
return result
# Background tasks
def send_notification_email(email: str, message: str):
"""Send notification email in background."""
# Email sending logic here
pass
@app.post("/users/{user_id}/notify")
async def notify_user(
user_id: int,
message: str,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_async_db)
):
"""Send notification to user."""
user = await get_user_by_id_async(db, user_id)
if not user:
raise UserNotFoundError(user_id)
background_tasks.add_task(send_notification_email, user.email, message)
return {"message": "Notification queued"}
🔧 Configuración y Variables de Entorno¶
Gestión de Configuración¶
from pydantic import BaseSettings, Field
from typing import Optional
class Settings(BaseSettings):
"""Application settings."""
# Database
database_url: str = Field(..., env="DATABASE_URL")
database_echo: bool = Field(False, env="DATABASE_ECHO")
# Security
secret_key: str = Field(..., env="SECRET_KEY")
algorithm: str = Field("HS256", env="ALGORITHM")
access_token_expire_minutes: int = Field(30, env="ACCESS_TOKEN_EXPIRE_MINUTES")
# API
api_v1_prefix: str = Field("/api/v1", env="API_V1_PREFIX")
debug: bool = Field(False, env="DEBUG")
# External services
redis_url: Optional[str] = Field(None, env="REDIS_URL")
email_service_url: Optional[str] = Field(None, env="EMAIL_SERVICE_URL")
class Config:
env_file = ".env"
case_sensitive = False
@lru_cache()
def get_settings() -> Settings:
"""Get cached settings instance."""
return Settings()
# Usage
settings = get_settings()
🐳 Containerización y Deployment¶
Dockerfile Optimizado¶
FROM python:3.11-slim
# Set environment variables
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PYTHONPATH=/app
# Set work directory
WORKDIR /app
# Install system dependencies
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
build-essential \
libpq-dev \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies
COPY requirements/prod.txt .
RUN pip install --no-cache-dir -r prod.txt
# Copy project
COPY . .
# Create non-root user
RUN adduser --disabled-password --gecos '' appuser && \
chown -R appuser:appuser /app
USER appuser
# Expose port
EXPOSE 8000
# Run application
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
📊 Logging y Monitoreo¶
Configuración de Logging¶
import logging
import sys
from datetime import datetime
from typing import Dict, Any
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler("app.log")
]
)
logger = logging.getLogger(__name__)
# Structured logging
def log_api_call(
method: str,
endpoint: str,
status_code: int,
duration: float,
user_id: Optional[int] = None
):
"""Log API call with structured data."""
log_data = {
"timestamp": datetime.utcnow().isoformat(),
"method": method,
"endpoint": endpoint,
"status_code": status_code,
"duration_ms": round(duration * 1000, 2),
"user_id": user_id
}
logger.info(f"API_CALL: {log_data}")
# Middleware for request logging
@app.middleware("http")
async def log_requests(request: Request, call_next):
"""Log all HTTP requests."""
start_time = time.time()
response = await call_next(request)
duration = time.time() - start_time
log_api_call(
method=request.method,
endpoint=str(request.url),
status_code=response.status_code,
duration=duration
)
return response
🔄 Patrones de Diseño¶
Repository Pattern¶
from abc import ABC, abstractmethod
from typing import List, Optional
from sqlalchemy.orm import Session
class BaseRepository(ABC):
"""Base repository interface."""
@abstractmethod
def get_by_id(self, id: int) -> Optional[Any]:
pass
@abstractmethod
def get_all(self, skip: int = 0, limit: int = 100) -> List[Any]:
pass
@abstractmethod
def create(self, obj_data: dict) -> Any:
pass
@abstractmethod
def update(self, id: int, obj_data: dict) -> Optional[Any]:
pass
@abstractmethod
def delete(self, id: int) -> bool:
pass
class UserRepository(BaseRepository):
"""User repository implementation."""
def __init__(self, db: Session):
self.db = db
def get_by_id(self, user_id: int) -> Optional[User]:
return self.db.query(User).filter(User.id == user_id).first()
def get_all(self, skip: int = 0, limit: int = 100) -> List[User]:
return self.db.query(User).offset(skip).limit(limit).all()
def create(self, user_data: dict) -> User:
user = User(**user_data)
self.db.add(user)
self.db.commit()
self.db.refresh(user)
return user
def update(self, user_id: int, user_data: dict) -> Optional[User]:
user = self.get_by_id(user_id)
if user:
for key, value in user_data.items():
setattr(user, key, value)
self.db.commit()
self.db.refresh(user)
return user
def delete(self, user_id: int) -> bool:
user = self.get_by_id(user_id)
if user:
self.db.delete(user)
self.db.commit()
return True
return False
Service Layer Pattern¶
class UserService:
"""User business logic service."""
def __init__(self, user_repository: UserRepository):
self.user_repository = user_repository
async def create_user(self, user_data: UserCreate) -> UserResponse:
"""Create a new user with business logic validation."""
# Business logic validation
if await self.email_exists(user_data.email):
raise ValidationError("Email already exists")
# Create user
user = self.user_repository.create(user_data.dict())
# Additional business logic (send welcome email, etc.)
await self.send_welcome_email(user.email)
return UserResponse.from_orm(user)
async def email_exists(self, email: str) -> bool:
"""Check if email already exists."""
user = self.user_repository.get_by_email(email)
return user is not None
async def send_welcome_email(self, email: str):
"""Send welcome email to new user."""
# Email sending logic here
pass
🎯 Guías Específicas de FastAPI¶
Dependency Injection¶
from fastapi import Depends
from sqlalchemy.orm import Session
# Database dependency
def get_db() -> Session:
"""Get database session."""
db = SessionLocal()
try:
yield db
finally:
db.close()
# Repository dependencies
def get_user_repository(db: Session = Depends(get_db)) -> UserRepository:
"""Get user repository instance."""
return UserRepository(db)
# Service dependencies
def get_user_service(
user_repo: UserRepository = Depends(get_user_repository)
) -> UserService:
"""Get user service instance."""
return UserService(user_repo)
# Usage in endpoints
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
user_service: UserService = Depends(get_user_service)
):
"""Get user by ID."""
return await user_service.get_user_by_id(user_id)
Middleware Personalizado¶
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
class TimingMiddleware(BaseHTTPMiddleware):
"""Middleware to measure request processing time."""
async def dispatch(self, request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
# Add middleware to app
app.add_middleware(TimingMiddleware)
💡 Consejo: Mantén siempre la consistencia en el estilo de código usando herramientas como Black, isort y flake8. Configura pre-commit hooks para automatizar estas verificaciones.