Saltar a contenido

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 typing cuando sea necesario
  • Usar Optional para valores que pueden ser None
  • Usar Union para 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-tools para 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.