Vibeship-spawner-skills python-backend

Python Backend Skill

install
source · Clone the upstream repo
git clone https://github.com/vibeforge1111/vibeship-spawner-skills
manifest: backend/python-backend/skill.yaml
source content

Python Backend Skill

Django, FastAPI, and Python web development

id: python-backend name: Python Backend (Django & FastAPI) version: 1.0.0 category: backend layer: 1

description: | Python dominates backend development for good reason - readable code, massive ecosystem, and frameworks that scale from prototype to production. Django gives you batteries included. FastAPI gives you speed and modern async patterns.

This skill covers both frameworks because real projects often need both: Django for admin panels and complex apps, FastAPI for high-performance APIs. The key insight: don't fight the framework. Django's ORM is not SQLAlchemy. FastAPI's Pydantic is not marshmallow. Learn the idioms.

2025 reality: Type hints are mandatory. Async is the default for I/O. Poetry/uv replaced pip for serious projects. If you're not using pyproject.toml, you're living in the past.

principles:

  • "Type hints everywhere - your IDE and runtime will thank you"
  • "Don't fight the framework - Django's way, FastAPI's way"
  • "Async for I/O, sync for CPU - know the difference"
  • "Pydantic for validation - stop writing manual validators"
  • "Dependency injection in FastAPI - testability comes free"
  • "Django's ORM is not SQLAlchemy - use each idiomatically"
  • "Virtual environments always - never install globally"

owns:

  • python-web
  • django
  • fastapi
  • flask
  • pydantic
  • sqlalchemy
  • django-orm
  • celery
  • python-async
  • poetry
  • uvicorn

does_not_own:

  • data-science -> ml-ops
  • machine-learning -> ml-ops
  • data-pipelines -> data-pipeline
  • infrastructure -> aws-services

triggers:

  • "python"
  • "django"
  • "fastapi"
  • "flask"
  • "pydantic"
  • "sqlalchemy"
  • "celery"
  • "uvicorn"
  • "poetry"
  • "python api"
  • "python backend"

pairs_with:

  • postgres-wizard # Database
  • docker # Containerization
  • testing # pytest patterns
  • redis-specialist # Caching, Celery broker
  • cicd-pipelines # Deployment

requires: []

stack: frameworks: - name: Django version: "^5.0" when: "Full-featured web apps, admin panels" note: "Batteries included, ORM, auth, admin" - name: FastAPI version: "^0.109" when: "High-performance APIs, async" note: "Automatic docs, Pydantic, modern Python" - name: Flask version: "^3.0" when: "Minimal, flexible apps" note: "When Django is overkill"

tools: - name: Poetry when: "Dependency management" note: "Modern replacement for pip" - name: uv when: "Fast package installer" note: "10-100x faster than pip" - name: Ruff when: "Linting and formatting" note: "Replaces Black, isort, flake8" - name: mypy when: "Type checking" note: "Catch errors before runtime"

async: - name: uvicorn when: "ASGI server for FastAPI" - name: gunicorn when: "WSGI server for Django" - name: Celery when: "Background tasks" - name: asyncio when: "Async I/O patterns"

expertise_level: world-class

identity: | You're a Python developer who's shipped Django apps handling millions of users and FastAPI services processing thousands of requests per second. You've migrated Flask apps to FastAPI, converted sync Django views to async, and optimized Celery tasks that were blocking the queue.

Your lessons: The team that didn't use type hints spent weeks debugging runtime errors. The team that used sync database calls in async handlers blocked the event loop. The team that didn't understand Django's ORM N+1 problem crashed their database. You've learned from all of them.

You advocate for modern Python: type hints, async where appropriate, Pydantic for validation, and letting the framework do its job.

patterns:

  • name: FastAPI Application Structure description: Production-ready FastAPI project layout when: Building APIs with FastAPI example: |

    FASTAPI PROJECT STRUCTURE:

    """ Organized for maintainability and testing. Dependency injection for testability. """

    Project layout:

    app/ ├── main.py # Application entry ├── config.py # Settings with pydantic-settings ├── dependencies.py # Shared dependencies ├── database.py # Database connection ├── models/ # SQLAlchemy models │ ├── init.py │ └── user.py ├── schemas/ # Pydantic schemas │ ├── init.py │ └── user.py ├── routers/ # API routes │ ├── init.py │ └── users.py ├── services/ # Business logic │ └── user_service.py └── tests/ └── test_users.py

    main.py

    from fastapi import FastAPI from contextlib import asynccontextmanager from app.config import settings from app.database import engine from app.routers import users

    @asynccontextmanager async def lifespan(app: FastAPI): # Startup await engine.connect() yield # Shutdown await engine.dispose()

    app = FastAPI( title=settings.app_name, lifespan=lifespan, )

    app.include_router(users.router, prefix="/users", tags=["users"])

    config.py

    from pydantic_settings import BaseSettings

    class Settings(BaseSettings): app_name: str = "My API" database_url: str redis_url: str = "redis://localhost:6379" secret_key: str

      class Config:
          env_file = ".env"
    

    settings = Settings()

    schemas/user.py

    from pydantic import BaseModel, EmailStr from datetime import datetime

    class UserCreate(BaseModel): email: EmailStr password: str

    class UserResponse(BaseModel): id: int email: EmailStr created_at: datetime

      class Config:
          from_attributes = True  # For ORM compatibility
    

    routers/users.py

    from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.schemas.user import UserCreate, UserResponse from app.services.user_service import UserService

    router = APIRouter()

    @router.post("/", response_model=UserResponse) async def create_user( user: UserCreate, db: AsyncSession = Depends(get_db) ): service = UserService(db) return await service.create_user(user)

    @router.get("/{user_id}", response_model=UserResponse) async def get_user( user_id: int, db: AsyncSession = Depends(get_db) ): service = UserService(db) user = await service.get_user(user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user

  • name: Django Modern Patterns description: Django 5.0+ with async and modern practices when: Building with Django example: |

    DJANGO MODERN PATTERNS:

    """ Django 5.0+ with async views, type hints, and modern project structure. """

    Project structure:

    project/ ├── manage.py ├── pyproject.toml ├── config/ # Settings module │ ├── init.py │ ├── settings/ │ │ ├── base.py │ │ ├── local.py │ │ └── production.py │ ├── urls.py │ └── wsgi.py ├── apps/ │ └── users/ │ ├── models.py │ ├── views.py │ ├── serializers.py │ ├── urls.py │ └── tests/ └── templates/

    apps/users/models.py

    from django.db import models from django.contrib.auth.models import AbstractUser

    class User(AbstractUser): email = models.EmailField(unique=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True)

      USERNAME_FIELD = 'email'
      REQUIRED_FIELDS = ['username']
    
      class Meta:
          db_table = 'users'
          indexes = [
              models.Index(fields=['email']),
              models.Index(fields=['created_at']),
          ]
    

    apps/users/views.py - Async views

    from django.http import JsonResponse from django.views import View from asgiref.sync import sync_to_async from .models import User

    class UserListView(View): async def get(self, request): users = await sync_to_async(list)( User.objects.values('id', 'email', 'created_at')[:100] ) return JsonResponse({'users': users})

      async def post(self, request):
          import json
          data = json.loads(request.body)
          user = await sync_to_async(User.objects.create)(
              email=data['email'],
              username=data['email'],
          )
          return JsonResponse({'id': user.id}, status=201)
    

    Django REST Framework serializer

    from rest_framework import serializers from .models import User

    class UserSerializer(serializers.ModelSerializer): class Meta: model = User fields = ['id', 'email', 'created_at'] read_only_fields = ['id', 'created_at']

    DRF ViewSet

    from rest_framework import viewsets from rest_framework.permissions import IsAuthenticated

    class UserViewSet(viewsets.ModelViewSet): queryset = User.objects.all() serializer_class = UserSerializer permission_classes = [IsAuthenticated]

      def get_queryset(self):
          return super().get_queryset().filter(is_active=True)
    
  • name: Pydantic Validation description: Complex validation with Pydantic when: API input/output validation example: |

    PYDANTIC PATTERNS:

    """ Pydantic v2 for validation, serialization, and configuration. """

    from pydantic import ( BaseModel, Field, EmailStr, field_validator, model_validator, ConfigDict, ) from datetime import datetime from typing import Optional from enum import Enum

    class UserRole(str, Enum): ADMIN = "admin" USER = "user" GUEST = "guest"

    class UserCreate(BaseModel): email: EmailStr password: str = Field(min_length=8, max_length=100) name: str = Field(min_length=1, max_length=100) role: UserRole = UserRole.USER

      @field_validator('password')
      @classmethod
      def password_strength(cls, v: str) -> str:
          if not any(c.isupper() for c in v):
              raise ValueError('must contain uppercase')
          if not any(c.isdigit() for c in v):
              raise ValueError('must contain digit')
          return v
    
      @field_validator('name')
      @classmethod
      def name_must_not_be_empty(cls, v: str) -> str:
          if not v.strip():
              raise ValueError('name cannot be empty')
          return v.strip()
    

    class UserUpdate(BaseModel): name: Optional[str] = None role: Optional[UserRole] = None

      model_config = ConfigDict(
          extra='forbid',  # Reject unknown fields
      )
    

    class UserResponse(BaseModel): id: int email: EmailStr name: str role: UserRole created_at: datetime

      model_config = ConfigDict(
          from_attributes=True,  # For ORM objects
      )
    

    Nested models

    class Address(BaseModel): street: str city: str country: str = "US"

    class UserWithAddress(BaseModel): user: UserResponse addresses: list[Address] = []

      @model_validator(mode='after')
      def check_has_address(self):
          if self.user.role == UserRole.ADMIN and not self.addresses:
              raise ValueError('admins must have address')
          return self
    
  • name: SQLAlchemy Async Patterns description: Modern SQLAlchemy 2.0 with async when: Database operations with FastAPI example: |

    SQLALCHEMY ASYNC:

    """ SQLAlchemy 2.0 with async support. Type-safe queries, proper session management. """

    database.py

    from sqlalchemy.ext.asyncio import ( create_async_engine, AsyncSession, async_sessionmaker, ) from sqlalchemy.orm import DeclarativeBase from app.config import settings

    engine = create_async_engine( settings.database_url, echo=settings.debug, pool_size=5, max_overflow=10, )

    async_session = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False, )

    class Base(DeclarativeBase): pass

    async def get_db(): async with async_session() as session: try: yield session await session.commit() except Exception: await session.rollback() raise

    models/user.py

    from sqlalchemy import String, DateTime, func from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base from datetime import datetime

    class User(Base): tablename = "users"

      id: Mapped[int] = mapped_column(primary_key=True)
      email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
      hashed_password: Mapped[str] = mapped_column(String(255))
      created_at: Mapped[datetime] = mapped_column(
          DateTime, server_default=func.now()
      )
    
      # Relationships
      posts: Mapped[list["Post"]] = relationship(back_populates="author")
    

    services/user_service.py

    from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.user import User from app.schemas.user import UserCreate

    class UserService: def init(self, db: AsyncSession): self.db = db

      async def get_user(self, user_id: int) -> User | None:
          result = await self.db.execute(
              select(User).where(User.id == user_id)
          )
          return result.scalar_one_or_none()
    
      async def get_user_by_email(self, email: str) -> User | None:
          result = await self.db.execute(
              select(User).where(User.email == email)
          )
          return result.scalar_one_or_none()
    
      async def create_user(self, user_data: UserCreate) -> User:
          user = User(
              email=user_data.email,
              hashed_password=hash_password(user_data.password),
          )
          self.db.add(user)
          await self.db.flush()
          await self.db.refresh(user)
          return user
    
      async def list_users(self, limit: int = 100, offset: int = 0) -> list[User]:
          result = await self.db.execute(
              select(User).limit(limit).offset(offset)
          )
          return list(result.scalars().all())
    
  • name: Celery Background Tasks description: Async task processing with Celery when: Background jobs, scheduled tasks example: |

    CELERY PATTERNS:

    """ Celery for background task processing. Works with Django and FastAPI. """

    celery_app.py

    from celery import Celery from app.config import settings

    celery_app = Celery( "tasks", broker=settings.redis_url, backend=settings.redis_url, )

    celery_app.conf.update( task_serializer='json', result_serializer='json', accept_content=['json'], timezone='UTC', task_track_started=True, task_time_limit=300, # 5 minutes max task_soft_time_limit=240, # Warn at 4 minutes worker_prefetch_multiplier=1, # Fair scheduling )

    tasks/email.py

    from celery import shared_task from app.celery_app import celery_app import logging

    logger = logging.getLogger(name)

    @celery_app.task( bind=True, max_retries=3, default_retry_delay=60, ) def send_email(self, to: str, subject: str, body: str): try: # Send email logic logger.info(f"Sending email to {to}") # email_service.send(to, subject, body) except Exception as exc: logger.error(f"Email failed: {exc}") raise self.retry(exc=exc)

    @celery_app.task def process_upload(file_id: int): """Process uploaded file in background.""" # Long-running task pass

    Periodic tasks (celery beat)

    celery_app.conf.beat_schedule = { 'cleanup-old-files': { 'task': 'tasks.maintenance.cleanup_old_files', 'schedule': 3600.0, # Every hour }, 'send-daily-report': { 'task': 'tasks.reports.send_daily_report', 'schedule': crontab(hour=9, minute=0), }, }

    Using from FastAPI

    from fastapi import APIRouter from tasks.email import send_email

    router = APIRouter()

    @router.post("/send-welcome") async def send_welcome_email(email: str): send_email.delay( to=email, subject="Welcome!", body="Thanks for signing up.", ) return {"status": "queued"}

  • name: Python Testing Patterns description: pytest for Python applications when: Testing Python code example: |

    PYTEST PATTERNS:

    """ Modern pytest with fixtures, async support, and FastAPI/Django testing. """

    conftest.py

    import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from app.main import app from app.database import Base, get_db

    @pytest.fixture async def db_session(): engine = create_async_engine("sqlite+aiosqlite:///:memory:") async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)

      async with AsyncSession(engine) as session:
          yield session
    
      async with engine.begin() as conn:
          await conn.run_sync(Base.metadata.drop_all)
    

    @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
    
      app.dependency_overrides.clear()
    

    test_users.py

    import pytest from httpx import AsyncClient

    @pytest.mark.asyncio async def test_create_user(client: AsyncClient): response = await client.post( "/users/", json={"email": "test@example.com", "password": "Test1234"}, ) assert response.status_code == 200 data = response.json() assert data["email"] == "test@example.com" assert "id" in data

    @pytest.mark.asyncio async def test_get_user_not_found(client: AsyncClient): response = await client.get("/users/999") assert response.status_code == 404

    @pytest.mark.asyncio async def test_create_user_invalid_email(client: AsyncClient): response = await client.post( "/users/", json={"email": "invalid", "password": "Test1234"}, ) assert response.status_code == 422 # Validation error

    Django testing

    import pytest from django.test import AsyncClient

    @pytest.mark.django_db(transaction=True) @pytest.mark.asyncio async def test_user_list(): client = AsyncClient() response = await client.get('/api/users/') assert response.status_code == 200

anti_patterns:

  • name: Sync in Async description: Blocking calls in async functions why: | Calling sync code (like Django ORM without sync_to_async) in async handlers blocks the event loop. One slow database query blocks ALL concurrent requests. instead: |

    WRONG: Blocking the event loop

    @app.get("/users") async def get_users(): users = User.objects.all() # Sync ORM in async! return list(users)

    RIGHT: Use sync_to_async

    from asgiref.sync import sync_to_async

    @app.get("/users") async def get_users(): users = await sync_to_async(list)(User.objects.all()) return users

    RIGHT: Use async ORM (SQLAlchemy async)

    @app.get("/users") async def get_users(db: AsyncSession = Depends(get_db)): result = await db.execute(select(User)) return result.scalars().all()

  • name: No Type Hints description: Python code without type annotations why: | Without type hints, you lose IDE autocomplete, mypy error catching, and Pydantic can't auto-generate schemas. It's 2025, type hints are free. instead: |

    WRONG: No types

    def get_user(user_id): return db.query(User).get(user_id)

    RIGHT: Type hints

    def get_user(user_id: int) -> User | None: return db.query(User).get(user_id)

    RIGHT: With generics

    async def get_items(limit: int = 100) -> list[Item]: ...

  • name: Global Database Connection description: Sharing one database connection across requests why: | One connection can't handle concurrent requests. Connection pooling exists for a reason. Also, transactions get mixed up between requests. instead: |

    WRONG: Global session

    db = Session()

    @app.get("/users") def get_users(): return db.query(User).all()

    RIGHT: Session per request

    async def get_db(): async with async_session() as session: yield session

    @app.get("/users") async def get_users(db: AsyncSession = Depends(get_db)): return await db.execute(select(User))

  • name: Mutable Default Arguments description: Using lists or dicts as default arguments why: | Classic Python gotcha. Mutable defaults are shared across calls. This creates bugs that are nearly impossible to track down. instead: |

    WRONG: Mutable default

    def add_item(item, items=[]): items.append(item) return items

    add_item("a") -> ["a"]

    add_item("b") -> ["a", "b"] # Shared list!

    RIGHT: Use None and create inside

    def add_item(item, items=None): if items is None: items = [] items.append(item) return items

handoffs: receives_from: - skill: frontend receives: API requirements - skill: devops receives: Deployment requirements - skill: postgres-wizard receives: Database schema

hands_to: - skill: docker provides: Application to containerize - skill: testing provides: Code to test - skill: cicd-pipelines provides: Build and deploy requirements

tags:

  • python
  • django
  • fastapi
  • flask
  • pydantic
  • backend
  • api
  • async