raheesahmed/sajicode/python-engineer

python -m venv .venv
source .venv/bin/activate  # Unix
.venv\Scripts\activate     # Windows
pip install fastapi uvicorn pydantic sqlalchemy alembic pytest httpx

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app + lifespan
│   ├── config.py             # Settings from env vars
│   ├── models/               # SQLAlchemy models
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/               # Pydantic schemas
│   │   └── user.py
│   ├── routes/                # API route handlers
│   │   └── users.py
│   ├── services/              # Business logic
│   │   └── user_service.py
│   ├── repositories/          # Database queries
│   │   └── user_repo.py
│   └── utils/                 # Helpers
│       └── auth.py
├── tests/
│   ├── conftest.py
│   └── test_users.py
├── alembic/
├── alembic.ini
├── pyproject.toml
├── requirements.txt
└── Dockerfile

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
@asynccontextmanager
async def lifespan(app: FastAPI):
    await database.connect()
    yield
    await database.disconnect()
app = FastAPI(title="My API", version="1.0.0", lifespan=lifespan)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

from fastapi import APIRouter, HTTPException, Depends, status
from pydantic import BaseModel, EmailStr, Field
router = APIRouter(prefix="/api/users", tags=["users"])
class CreateUserRequest(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=100)
    password: str = Field(min_length=8, max_length=128)
class UserResponse(BaseModel):
    id: str
    email: str
    name: str
    model_config = {"from_attributes": True}
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(data: CreateUserRequest, db: Session = Depends(get_db)):
    existing = await user_service.get_by_email(db, data.email)
    if existing:
        raise HTTPException(status_code=409, detail="Email already registered")
    user = await user_service.create(db, data)
    return user
@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: str, db: Session = Depends(get_db)):
    user = await user_service.get_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

from functools import lru_cache
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
    database_url: str
    jwt_secret: str
    debug: bool = False
    model_config = {"env_file": ".env"}
@lru_cache
def get_settings() -> Settings:
    return Settings()
async def get_db():
    async with AsyncSession(engine) as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
@pytest.mark.anyio
async def test_create_user(client: AsyncClient):
    response = await client.post("/api/users/", json={
        "email": "test@example.com",
        "name": "Test User",
        "password": "SecurePass123",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"
@pytest.mark.anyio
async def test_duplicate_email_returns_409(client: AsyncClient):
    payload = {"email": "dup@test.com", "name": "User", "password": "SecurePass123"}
    await client.post("/api/users/", json=payload)
    response = await client.post("/api/users/", json=payload)
    assert response.status_code == 409

import typer
from rich.console import Console
from rich.table import Table
app = typer.Typer(help="My CLI tool")
console = Console()
@app.command()
def greet(name: str, count: int = typer.Option(1, help="Number of greetings")):
    for _ in range(count):
        console.print(f"Hello, [bold magenta]{name}[/bold magenta]!")
@app.command()
def list_users():
    table = Table(title="Users")
    table.add_column("ID", style="dim")
    table.add_column("Name", style="bold")
    table.add_column("Email")
    for user in get_users():
        table.add_row(user.id, user.name, user.email)
    console.print(table)
if __name__ == "__main__":
    app()

import pandas as pd
def process_csv(input_path: str, output_path: str) -> None:
    df = pd.read_csv(input_path)
    df.columns = df.columns.str.strip().str.lower().str.replace(" ", "_")
    df = df.dropna(subset=["email"])
    df["email"] = df["email"].str.lower().str.strip()
    df = df.drop_duplicates(subset=["email"], keep="last")
    df.to_csv(output_path, index=False)