1# SQL Database — Python — REFERENCE ONLY
2 
3## SQLAlchemy Setup
4 
5Add SQLAlchemy with Azure SQL and managed identity support to a FastAPI app.
6 
7### Requirements
8 
9Add to `requirements.txt`:
10 
11```
12sqlalchemy>=2.0
13pyodbc
14azure-identity
15fastapi
16uvicorn
17```
18 
19### Database Configuration
20 
21Create `database.py`:
22 
23```python
24import os
25import struct
26from sqlalchemy import create_engine, event
27from sqlalchemy.orm import sessionmaker, DeclarativeBase
28from azure.identity import ManagedIdentityCredential
29 
30 
31class Base(DeclarativeBase):
32    pass
33 
34 
35def create_db_engine():
36    """Create SQLAlchemy engine using managed identity or connection string."""
37    conn_str = os.environ.get("AZURE_SQL_CONNECTION_STRING")
38    if conn_str:
39        # Local dev: AZURE_SQL_CONNECTION_STRING is a SQLAlchemy URL
40        # e.g. mssql+pyodbc://sa:password@localhost/myapp?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes
41        return create_engine(conn_str)
42 
43    # Azure: use managed identity token (refreshed per-connection so pool stays valid past 1 hour)
44    server = os.environ["AZURE_SQL_SERVER"]
45    database = os.environ["AZURE_SQL_DATABASE"]
46    client_id = os.environ.get("AZURE_CLIENT_ID")
47    credential = ManagedIdentityCredential(client_id=client_id) if client_id else ManagedIdentityCredential()
48 
49    odbc_conn = (
50        f"DRIVER={{ODBC Driver 18 for SQL Server}};"
51        f"SERVER={server};"
52        f"DATABASE={database};"
53        f"Encrypt=yes;TrustServerCertificate=no;"
54    )
55    engine = create_engine(
56        "mssql+pyodbc://",
57        connect_args={"odbc_connect": odbc_conn},
58    )
59 
60    @event.listens_for(engine, "do_connect")
61    def provide_token(dialect, conn_rec, cargs, cparams):
62        # Fetch a fresh token for every new physical connection — pool refills won't fail after 1h
63        token = credential.get_token("https://database.windows.net/.default")
64        token_bytes = token.token.encode("utf-16-le")
65        token_struct = struct.pack(f"<I{len(token_bytes)}s", len(token_bytes), token_bytes)
66        cparams["attrs_before"] = {1256: token_struct}  # SQL_COPT_SS_ACCESS_TOKEN
67 
68    return engine
69 
70 
71engine = create_db_engine()
72SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
73```
74 
75### Models
76 
77Create `models.py`:
78 
79```python
80from sqlalchemy import Column, Integer, String, Boolean
81from database import Base
82 
83class TodoItem(Base):
84    __tablename__ = "todo_items"
85 
86    id = Column(Integer, primary_key=True, index=True)
87    title = Column(String(200), nullable=False)
88    is_complete = Column(Boolean, default=False)
89```
90 
91### API Endpoints
92 
93Add to `main.py` — do NOT replace existing routes:
94 
95```python
96from fastapi import Depends, HTTPException
97from sqlalchemy.orm import Session
98from database import SessionLocal, engine, Base
99from models import TodoItem
100from pydantic import BaseModel
101 
102Base.metadata.create_all(bind=engine)
103 
104def get_db():
105    db = SessionLocal()
106    try:
107        yield db
108    finally:
109        db.close()
110 
111class TodoCreate(BaseModel):
112    title: str
113    is_complete: bool = False
114 
115@app.get("/api/todos")
116def list_todos(db: Session = Depends(get_db)):
117    return db.query(TodoItem).all()
118 
119@app.get("/api/todos/{todo_id}")
120def get_todo(todo_id: int, db: Session = Depends(get_db)):
121    todo = db.query(TodoItem).filter(TodoItem.id == todo_id).first()
122    if not todo:
123        raise HTTPException(status_code=404, detail="Not found")
124    return todo
125 
126@app.post("/api/todos", status_code=201)
127def create_todo(todo: TodoCreate, db: Session = Depends(get_db)):
128    db_todo = TodoItem(**todo.model_dump())
129    db.add(db_todo)
130    db.commit()
131    db.refresh(db_todo)
132    return db_todo
133```
134 
135### Local Development
136 
137For local development without managed identity, use SQL authentication:
138 
139```python
140# .env (local only — never commit)
141AZURE_SQL_CONNECTION_STRING=mssql+pyodbc://<username>:<password>@localhost/myapp?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes
142```
143 
144## Files to Add
145 
146| File | Action |
147|------|--------|
148| `database.py` | Create — engine, session, managed identity token |
149| `models.py` | Create — SQLAlchemy ORM models |
150| `main.py` | Modify — add CRUD endpoints + DB dependency |
151| `requirements.txt` | Modify — add sqlalchemy, pyodbc, azure-identity |
152 
153## Common Patterns
154 
155- Use `Depends(get_db)` for session lifecycle management
156- Use `Base.metadata.create_all()` only for dev; use Alembic migrations in production
157- Never store SQL passwords in app settings — use managed identity tokens
158