Backend Quick Start Guide

For Developers New to FastAPI

This guide helps you quickly understand how to work with the LCO backend, especially if you're coming from a frontend background or are new to FastAPI.

---

Basic Concepts

What is FastAPI?

FastAPI is a modern Python web framework for building APIs. Think of it like Express.js for Node.js, but with: - Automatic data validation - Auto-generated API documentation - Type safety through Python type hints - Built-in async support

Key Differences from Frontend Development

Frontend (React/TypeScript)	Backend (FastAPI/Python)
Components render UI	Routes handle HTTP requests
State management (Redux/Context)	Database queries
onClick handlers	@router.post() decorators
fetch() API calls	Router endpoints
TypeScript interfaces	Pydantic models
camelCase	snake_case

---

5-Minute Setup

1. Prerequisites

# Python 3.12 or higher recommended (3.10+ minimum)
python --version  # Should be 3.10 or higher

# pip package manager (comes with Python)
pip --version

# Optional: Check if you have Python 3.12 features available
python -c "print('Python 3.12+ type hints supported')" 2>/dev/null

2. Setup

# Navigate to backend directory
cd backend

# Create virtual environment (highly recommended)
python -m venv .venv

# Activate virtual environment
# On macOS/Linux:
source .venv/bin/activate
# On Windows PowerShell:
.venv\Scripts\Activate.ps1
# On Windows Command Prompt:
.venv\Scripts\activate.bat

# Upgrade pip to latest version
pip install --upgrade pip

# Install all dependencies
pip install -r requirements.txt

# Copy environment template
cp .env.example .env
# Edit .env with your Azure Cosmos DB credentials:
# - COSMOS_ENDPOINT (required)
# - COSMOS_KEY (required)
# - DATABASE_NAME (default: lco-construction)

3. Run

# Start development server with auto-reload
uvicorn app.main:app --reload --port 8000

# Alternative: Use the npm script from project root
# npm run dev:backend

# Server starts at http://localhost:8000
# Interactive API docs: http://localhost:8000/api/docs
# Alternative docs: http://localhost:8000/api/redoc
# Health check: http://localhost:8000/api/v1/health

---

Understanding the Code Flow

Example: Creating a Client

1. Frontend sends request:

// Frontend (TypeScript)
const response = await fetch('http://localhost:8000/api/v1/clients', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    clientCode: 'ACME',
    clientName: 'ACME Corporation',
    clientType: 'private',
    industry: 'commercial'
  })
});

2. Router receives request:

# Backend: app/api/v1/routers/clients.py
@router.post("", response_model=Client, status_code=status.HTTP_201_CREATED)
async def create_client(
    client_data: ClientCreate,  # ← FastAPI validates against this schema
    repo: ClientRepository = Depends(get_client_repo)  # ← Dependency injection
):
    # Check if client already exists
    existing = await repo.get_client_by_code(client_data.client_code)
    if existing:
        raise HTTPException(status_code=409, detail="Client already exists")

    # Create the client
    client = await repo.create_client(client_data)
    return client  # ← FastAPI serializes to JSON

3. Repository saves to database:

# Backend: app/repositories/client.py
async def create_client(self, client_data: ClientCreate) -> Client:
    # Convert Pydantic model to dict
    client_dict = client_data.model_dump(by_alias=True, mode='json')

    # Add metadata
    client_dict['type'] = 'client'
    client_dict['clientId'] = client_dict.get('id', client_dict.get('clientCode'))

    # Save to Cosmos DB
    result = await self.create(client_dict)  # ← Calls BaseRepository.create()

    # Convert back to Pydantic model
    return Client(**result)

4. Response flows back:

Cosmos DB → Repository → Router → Frontend

---

Common Tasks

Adding a New Field to an Existing Entity

Example: Add website field to Client

Step 1: Update the schema

# app/schemas/client.py
class ClientBase(BaseSchema):
    client_code: str = Field(alias="clientCode")
    client_name: str = Field(alias="clientName")
    client_type: ClientType = Field(alias="clientType")
    website: Optional[str] = None  # ← Add new field

Step 2: That's it! - Pydantic automatically handles validation - FastAPI updates the API documentation - No database migration needed (Cosmos DB is schema-less)

Adding a New Endpoint

Example: Get clients by industry

# app/api/v1/routers/clients.py

@router.get("/by-industry/{industry}", response_model=PaginatedResponse)
async def get_clients_by_industry(
    industry: str,  # ← Path parameter
    skip: int = Query(0, ge=0),  # ← Query parameter with validation
    limit: int = Query(20, ge=1, le=100),
    repo: ClientRepository = Depends(get_client_repo)
):
    """Get all clients in a specific industry"""
    result = await repo.get_all_clients(
        industry=industry,
        skip=skip,
        limit=limit
    )
    return PaginatedResponse(**result)

URL: GET /api/v1/clients/by-industry/commercial?skip=0&limit=20

Adding a Custom Repository Method

Example: Get top clients by project count

# app/repositories/client.py

async def get_top_clients(self, limit: int = 10) -> List[Client]:
    """Get clients with most projects"""
    query = """
    SELECT * FROM c
    WHERE c.type = @type
    ORDER BY c.projectCount DESC
    """
    parameters = [{"name": "@type", "value": "client"}]

    results = await self.query(
        query=query,
        parameters=parameters,
        max_item_count=limit
    )

    return [Client(**r) for r in results]

Handling Errors Gracefully

@router.get("/{client_id}", response_model=Client)
async def get_client(
    client_id: str,
    repo: ClientRepository = Depends(get_client_repo)
):
    try:
        client = await repo.get_client(client_id)

        if not client:
            # Return 404 if not found
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Client not found"
            )

        return client

    except HTTPException:
        # Re-raise HTTP exceptions as-is
        raise
    except Exception as e:
        # Log unexpected errors
        logger.error(f"Failed to get client {client_id}: {str(e)}")
        # Return 500 for unexpected errors
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to retrieve client"
        )

---

Understanding Key Concepts

1. Decorators

In Python, decorators are functions that modify other functions. FastAPI uses them heavily:

@router.get("/clients")  # ← Decorator
async def get_clients():  # ← Function being decorated
    pass

# Equivalent to:
# get_clients = router.get("/clients")(get_clients)

Common FastAPI decorators: - @router.get() - Handle GET requests - @router.post() - Handle POST requests - @router.put() - Handle PUT requests - @router.delete() - Handle DELETE requests

2. Type Hints

Python type hints provide type safety (like TypeScript):

# Without type hints
def add(a, b):
    return a + b

# With type hints
def add(a: int, b: int) -> int:
    return a + b

FastAPI uses type hints for: - Request validation - Response serialization - API documentation - IDE autocomplete

3. Async/Await

Similar to JavaScript async/await:

// TypeScript
async function getUser(id: string): Promise<User> {
  const response = await fetch(`/api/users/${id}`);
  return await response.json();
}

# Python
async def get_user(user_id: str) -> User:
    result = await container.read_item(item=user_id, partition_key=user_id)
    return User(**result)

Rules: - Use async def for async functions - Use await when calling async functions - Always await database operations

4. Dependency Injection

FastAPI's dependency injection is like React Context, but for the backend:

# Define a dependency (like a Context Provider)
async def get_current_user(token: str = Header(...)):
    # Validate token, return user
    return user

# Use the dependency (like useContext)
@router.get("/me")
async def get_me(user: User = Depends(get_current_user)):
    return user

# FastAPI automatically:
# 1. Calls get_current_user()
# 2. Passes result as 'user' parameter

5. Pydantic Models

Pydantic models are like TypeScript interfaces with validation:

// TypeScript interface (no validation)
interface Client {
  clientCode: string;
  clientName: string;
  clientType: 'private' | 'government';
}

# Pydantic model (with validation)
class Client(BaseModel):
    client_code: str = Field(alias="clientCode")
    client_name: str = Field(alias="clientName")
    client_type: Literal['private', 'government'] = Field(alias="clientType")

# Auto-validates when created
client = Client(clientCode="ACME", clientName="ACME Corp", clientType="private")
# Raises error if invalid
client = Client(clientCode="ACME")  # ← Error: missing required field

---

Project Structure Map

backend/app/
├── main.py                    # START HERE - FastAPI app with lifespan management
│   ↓ (initializes database on startup)
├── core/
│   └── config.py             # Pydantic Settings (environment variables)
│   ↓
├── db/
│   └── cosmos.py             # Async Cosmos DB client (connection pooling)
│   ↓
├── api/v1/routers/           # FastAPI routers (HTTP endpoints)
│   ├── clients.py            # Client CRUD endpoints
│   ├── projects.py           # Project CRUD endpoints
│   ├── services.py           # Service endpoints (nested under projects)
│   ├── crews.py              # Crew, CrewTrade, CrewMember endpoints
│   ├── service_crews.py      # Service crew assignment endpoints
│   ├── equipment.py          # Equipment catalog endpoints
│   └── mto.py                # Material Takeoff endpoints
│   ↓ (use dependency injection)
├── repositories/              # Repository pattern (data access layer)
│   ├── base.py               # BaseRepository[T] with generic CRUD
│   ├── client.py             # ClientRepository (extends base)
│   ├── project.py            # ProjectRepository
│   ├── crew.py               # CrewRepository, CrewTradeRepository, CrewMemberRepository
│   ├── service.py            # ServiceRepository
│   ├── service_crew.py       # ServiceCrewRepository
│   ├── equipment.py          # EquipmentRepository
│   └── mto.py                # MaterialTakeoffRepository
│   ↓ (use Pydantic models)
└── schemas/                   # Pydantic models (validation + serialization)
    ├── base.py               # BaseSchema, BaseDocument, PaginatedResponse
    ├── common.py             # Shared models
    ├── client.py             # Client, ClientCreate, ClientUpdate
    ├── project.py            # Project models
    ├── crew.py               # Crew models (3-tier system)
    ├── service.py            # Service models
    ├── service_crew.py       # ServiceCrew with indirect costs
    ├── equipment.py          # Equipment models
    └── mto.py                # MaterialTakeoff models

Mental model for frontend developers: - schemas = TypeScript interfaces (with runtime validation) - routers = Express routes (with automatic OpenAPI generation) - repositories = Database service layer (with async/await) - main.py = Express app setup (with middleware and lifespan hooks)

---

FastAPI vs Frontend Patterns

Making API Calls

Frontend (calling backend):

const client = await fetch('/api/v1/clients/123').then(r => r.json());

Backend (querying database):

client = await repo.get_client('123')

Validation

Frontend:

// Manual validation
if (!clientName || clientName.length < 3) {
  throw new Error('Client name too short');
}

Backend:

# Automatic validation via Pydantic
class ClientCreate(BaseModel):
    client_name: str = Field(min_length=3, alias="clientName")
    # Raises validation error automatically

Error Handling

Frontend:

try {
  const data = await fetchData();
} catch (error) {
  console.error(error);
  showErrorToast();
}

Backend:

try:
    data = await repo.get_data()
except ValueError as e:
    raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
    logger.error(f"Error: {str(e)}")
    raise HTTPException(status_code=500, detail="Internal error")

---

Testing Your Changes

1. Use the Interactive API Docs

1. Start the server: uvicorn app.main:app --reload
2. Open http://localhost:8000/api/docs
3. Click on any endpoint
4. Click "Try it out"
5. Fill in parameters
6. Click "Execute"
7. See the response

2. Use curl

# GET request
curl http://localhost:8000/api/v1/clients

# POST request
curl -X POST http://localhost:8000/api/v1/clients \
  -H "Content-Type: application/json" \
  -d '{
    "clientCode": "ACME",
    "clientName": "ACME Corp",
    "clientType": "private"
  }'

3. Use Python REPL

# Start Python shell
python

# Import and test
>>> from app.repositories.client import ClientRepository
>>> from app.db.cosmos import cosmos_db
>>> import asyncio

>>> async def test():
...     await cosmos_db.initialize()
...     repo = ClientRepository(cosmos_db.get_container("clients"))
...     client = await repo.get_client("some-id")
...     print(client)

>>> asyncio.run(test())

---

Common Errors & Solutions

Error: "Field required"

Problem: Missing required field in request

class ClientCreate(BaseModel):
    client_code: str  # ← Required
    client_name: str  # ← Required

Solution: Send all required fields from frontend

Error: "Cosmos DB not initialized"

Problem: Database connection not established

cosmos_db.get_container("clients")  # ← Called before initialize()

Solution: Ensure await cosmos_db.initialize() is called in startup

Error: "Container not found"

Problem: Wrong container name

cosmos_db.get_container("Clients")  # ← Case-sensitive!

Solution: Use exact name from config.py:

cosmos_db.get_container("clients")  # ← Correct

Error: "Partition key mismatch"

Problem: Partition key doesn't match container configuration

# Container expects /clientId
await container.read_item(item=id, partition_key="wrong-key")

Solution: Use correct partition key:

await container.read_item(item=id, partition_key=client_id)

---

Best Practices

1. Always Use Type Hints

# Bad
async def get_client(id):
    return await repo.get(id)

# Good
async def get_client(client_id: str) -> Optional[Client]:
    return await repo.get_client(client_id)

2. Use Pydantic Models for Validation

# Bad - manual validation
if not client_name or len(client_name) < 3:
    raise ValueError("Invalid name")

# Good - Pydantic handles it
class ClientCreate(BaseModel):
    client_name: str = Field(min_length=3, alias="clientName")

3. Handle Errors Properly

# Bad
@router.get("/{id}")
async def get_client(id: str):
    return await repo.get_client(id)  # ← What if it fails?

# Good
@router.get("/{id}")
async def get_client(client_id: str, repo = Depends(get_repo)):
    try:
        client = await repo.get_client(client_id)
        if not client:
            raise HTTPException(status_code=404, detail="Not found")
        return client
    except Exception as e:
        logger.error(f"Error: {e}")
        raise HTTPException(status_code=500, detail="Internal error")

4. Use Dependency Injection

# Bad - creating repo in each endpoint
@router.get("/")
async def get_clients():
    repo = ClientRepository(cosmos_db.get_container("clients"))
    return await repo.get_all_clients()

# Good - inject repo
async def get_repo():
    return ClientRepository(cosmos_db.get_container("clients"))

@router.get("/")
async def get_clients(repo: ClientRepository = Depends(get_repo)):
    return await repo.get_all_clients()

5. Always Await Async Functions

# Bad
client = repo.get_client(id)  # ← Missing await!

# Good
client = await repo.get_client(id)

---

Helpful Commands

# Start development server with auto-reload
uvicorn app.main:app --reload --port 8000

# Start with custom host (accessible from other devices)
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# Check Python version
python --version

# Install new package
pip install package-name
pip freeze > requirements.txt  # Update requirements

# Activate virtual environment
source .venv/bin/activate  # macOS/Linux
.venv\Scripts\activate     # Windows

# Deactivate virtual environment
deactivate

---

Async Operations Example

For long-running operations (e.g., MTO import), use background tasks:

# Submit async task
@router.post("/import/async", response_model=TaskSubmitResponse)
async def import_async(
    background_tasks: BackgroundTasks,
    file: UploadFile,
    project_id: str = Form(...)
):
    # Create task record
    task = await task_service.create_task("mto_import", project_id)

    # Queue background job
    background_tasks.add_task(process_import, task["taskId"], file_content)

    # Return task ID immediately
    return TaskSubmitResponse(
        taskId=task["taskId"],
        status="pending",
        pollUrl=f"/api/v1/import/status/{task['taskId']}"
    )

# Poll status
@router.get("/import/status/{task_id}")
async def get_status(task_id: str):
    task = await task_service.get_task(task_id)
    return TaskResponse(**task)

See MTO Import/Export Guide for complete async workflow documentation.

---

Next Steps

1. Read the full architecture guide: docs/backend/ARCHITECTURE.md
2. Read the MTO Import/Export Guide: docs/backend/MTO_IMPORT_EXPORT_GUIDE.md (comprehensive async import documentation)
3. Explore the Swagger UI: http://localhost:8000/api/docs
4. Try modifying an endpoint: Start with clients.py
5. Add a new field: Update a schema and see it work
6. Write a custom query: Add a method to a repository

---

Resources

• FastAPI Tutorial: https://fastapi.tiangolo.com/tutorial/
• Pydantic Docs: https://docs.pydantic.dev/
• Python Async/Await: https://realpython.com/async-io-python/
• Our Architecture Guide: docs/backend/ARCHITECTURE.md

---

Pro Tip: The Swagger UI (/api/docs) is your best friend. It's automatically generated from your code and lets you test endpoints interactively!