Python Backend Development Skill

import pytest
from my_service import UserService, UserNotFoundError

class TestUserService:
    @pytest.mark.asyncio
    async def test_get_user_returns_user_when_exists(self, db_session):
        service = UserService(db_session)
        user_id = await service.create_user("alice", "alice@example.com")
        user = await service.get_user(user_id)
        assert user.username == "alice"

    @pytest.mark.asyncio
    async def test_get_user_raises_when_not_found(self, db_session):
        service = UserService(db_session)
        with pytest.raises(UserNotFoundError):
            await service.get_user(99999)

    @pytest.mark.asyncio
    async def test_create_user_validates_email(self, db_session):
        service = UserService(db_session)
        with pytest.raises(ValueError, match="Invalid email"):
            await service.create_user("bob", "not-an-email")

class UserNotFoundError(Exception): pass

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

    async def get_user(self, user_id: int) -> User:
        user = await self.db.get(User, user_id)
        if not user:
            raise UserNotFoundError(f"User {user_id} not found")
        return user

    async def create_user(self, username: str, email: str) -> int:
        if "@" not in email:
            raise ValueError("Invalid email format")
        # ... minimal implementation to pass tests

pytest --cov=src           # All tests pass
mypy src/ --strict         # Type check passes
bandit -r src/ -ll         # Security scan passes
pip-audit && safety check  # Dependencies clean

# BAD: Sequential requests (slow)
for url in urls:
    response = await client.get(url)  # Waits for each one

# GOOD: Concurrent requests with gather
tasks = [client.get(url) for url in urls]
responses = await asyncio.gather(*tasks)  # All at once

# BAD: Load all into memory
return [process(line) for line in f.readlines()]  # OOM risk

# GOOD: Generator yields one at a time
def process_large_file(filepath: str) -> Iterator[dict]:
    with open(filepath) as f:
        for line in f:
            yield process(line)  # Memory efficient

# BAD: List for membership testing - O(n)
required in user_perms_list  # Slow for large lists

# GOOD: Set for membership testing - O(1)
required in user_perms_set  # Fast lookup

# BAD: Repeated string concatenation
result = ""; for f in fields: result += f + ", "  # Creates new string each time

# GOOD: Join for string building
", ".join(fields)  # Single allocation

# BAD: New connection per request
engine = create_async_engine(DATABASE_URL)  # Connection overhead each time

# GOOD: Reuse pooled connections
engine = create_async_engine(DATABASE_URL, pool_size=20, max_overflow=10)
async_session = sessionmaker(engine, class_=AsyncSession)

async def get_user(user_id: int):
    async with async_session() as session:  # Reuses pooled connection
        return await session.get(User, user_id)

# BAD: Individual inserts (N round trips)
for user in users:
    db.add(User(**user)); await db.commit()  # N commits = slow

# GOOD: Batch insert (1 round trip)
stmt = insert(User).values(users)
await db.execute(stmt); await db.commit()  # Single commit

# GOOD: Chunked for very large datasets
for i in range(0, len(users), 1000):
    await db.execute(insert(User).values(users[i:i+1000]))
await db.commit()

# pyproject.toml
[project]
dependencies = [
    "pydantic>=2.0", "email-validator>=2.0",      # Validation
    "cryptography>=41.0", "argon2-cffi>=21.0",    # Cryptography
    "PyJWT>=2.8", "sqlalchemy>=2.0", "asyncpg>=0.28",
    "httpx>=0.25", "bandit>=1.7",
]

[project.optional-dependencies]
dev = ["pytest>=7.0", "pytest-asyncio>=0.21", "hypothesis>=6.0", "safety>=2.0", "pip-audit>=2.0"]

from pydantic import BaseModel, Field, field_validator, EmailStr
from typing import Annotated
import re

class UserCreate(BaseModel):
    """Validated user creation request."""
    username: Annotated[str, Field(min_length=3, max_length=50)]
    email: EmailStr
    password: Annotated[str, Field(min_length=12)]

    @field_validator('username')
    @classmethod
    def validate_username(cls, v: str) -> str:
        if not re.match(r'^[a-zA-Z0-9_-]+$', v):
            raise ValueError('Username must be alphanumeric')
        return v

    @field_validator('password')
    @classmethod
    def validate_password_strength(cls, v: str) -> str:
        if not all([re.search(r'[A-Z]', v), re.search(r'[a-z]', v), re.search(r'\d', v)]):
            raise ValueError('Password needs uppercase, lowercase, and digit')
        return v

from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError

ph = PasswordHasher(time_cost=3, memory_cost=65536, parallelism=4)

def hash_password(password: str) -> str:
    return ph.hash(password)

def verify_password(password: str, hash: str) -> bool:
    try:
        ph.verify(hash, password)
        return True
    except VerifyMismatchError:
        return False

from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

# NEVER: f"SELECT * FROM users WHERE username = '{username}'"

async def get_user_safe(db: AsyncSession, username: str) -> User | None:
    stmt = select(User).where(User.username == username)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

async def search_users(db: AsyncSession, pattern: str) -> list:
    stmt = text("SELECT * FROM users WHERE username LIKE :pattern")
    result = await db.execute(stmt, {"pattern": f"%{pattern}%"})
    return result.fetchall()

from pathlib import Path

def safe_read_file(base_dir: Path, user_filename: str) -> str:
    if '..' in user_filename or user_filename.startswith('/'):
        raise ValueError("Invalid filename")

    file_path = (base_dir / user_filename).resolve()
    if not file_path.is_relative_to(base_dir.resolve()):
        raise ValueError("Path traversal detected")

    return file_path.read_text()

import subprocess

ALLOWED_PROGRAMS = {'git', 'python', 'pip'}

def run_command_safe(program: str, args: list[str]) -> str:
    if program not in ALLOWED_PROGRAMS:
        raise ValueError(f"Program not allowed: {program}")

    result = subprocess.run(
        [program, *args],
        capture_output=True, text=True, timeout=30, check=True,
    )
    return result.stdout

from pydantic import BaseModel, field_validator
import os, logging

# Secure base model - reject unknown fields, strip whitespace
class SecureInput(BaseModel):
    model_config = {'extra': 'forbid', 'str_strip_whitespace': True}

    @field_validator('*', mode='before')
    @classmethod
    def reject_null_bytes(cls, v):
        if isinstance(v, str) and '\x00' in v:
            raise ValueError('Null bytes not allowed')
        return v

# Secrets from environment (NEVER hardcode)
API_KEY = os.environ["API_KEY"]
DB_URL = os.environ["DATABASE_URL"]

# Safe error handling - log details, return safe message
class AppError(Exception):
    def __init__(self, message: str, internal: str = None):
        self.message = message
        if internal:
            logging.error(f"{message}: {internal}")

    def to_response(self) -> dict:
        return {"error": self.message}

bandit -r src/ -ll          # Static analysis
pip-audit && safety check   # Dependency vulnerabilities
mypy src/ --strict          # Type checking

import pytest
from pathlib import Path

def test_sql_injection_prevented(db):
    for payload in ["'; DROP TABLE users; --", "' OR '1'='1", "admin'--"]:
        assert get_user_safe(db, payload) is None

def test_path_traversal_blocked():
    base = Path("/app/data")
    for attack in ["../etc/passwd", "..\\windows\\system32", "foo/../../etc/passwd"]:
        with pytest.raises(ValueError, match="traversal|Invalid"):
            safe_read_file(base, attack)

def test_command_injection_blocked():
    with pytest.raises(ValueError, match="not allowed"):
        run_command_safe("rm", ["-rf", "/"])