Core Packages for Fast API

Core Packages for Fast API

Core Packages for Fast API

Exploring more powerful and common packages and their use cases beyond the core built-in features.

1. Configuration ⚙️

Use Case: Managing application settings, secrets, and environment variables. Using Pydantic's Settings (from pydantic-settings) and python-dotenv is the standard for creating a robust and type-safe configuration. It allows you to define settings in a class and automatically load values from environment variables, a .env file, or other sources.

Python

# .env file
# This file is not checked into version control
DATABASE_URL="postgresql://user:password@host:port/dbname"
SECRET_KEY="your-secret-key-for-jwt"

# config.py (a new file)
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str

    class Config:
        env_file = ".env"

# main.py
from fastapi import FastAPI
from .config import Settings

settings = Settings()
app = FastAPI()

@app.get("/config")
def read_config():
    return {"database_url": settings.database_url}

2. Database Integration 💾

2.a SQL (Relational Databases)

Use Case: Connecting to and managing data in a relational database. Using an Object-Relational Mapper (ORM) like SQLAlchemy, or the more modern SQLModel, is a best practice. Alembic is the standard tool for managing database schema migrations.

Python (with SQLModel)

# main.py (continued from above)
from sqlmodel import SQLModel, Field, Session, create_engine
from fastapi import Depends, HTTPException

# Assuming settings is configured from the previous section
engine = create_engine(settings.database_url)

# Define a model that is both a Pydantic model and a database table
class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: int | None = None

def get_db():
    with Session(engine) as session:
        yield session

@app.post("/heroes/")
def create_hero(*, session: Session = Depends(get_db), hero: Hero):
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero

2.b NoSQL (Document Databases)

Use Case: Working with document-based databases for flexible, scalable data storage. For MongoDB, the motor asynchronous driver is the standard. You can still use Pydantic for validation and serialization, which works well with the flexible schema of NoSQL.

Python

# main.py (continued)
from fastapi import Depends, FastAPI
from pydantic import BaseModel
import motor.motor_asyncio

# Assuming settings is configured from the previous section
client = motor.motor_asyncio.AsyncIOMotorClient(settings.database_url)
db = client["fastapi_db"]

class Product(BaseModel):
    name: str
    price: float

async def get_db():
    return db

@app.post("/products/")
async def create_product(product: Product, database=Depends(get_db)):
    product_dict = product.model_dump()
    result = await database["products"].insert_one(product_dict)
    return {"id": str(result.inserted_id)}

3. Authentication & Authorization 🔒

Use Case: Securing endpoints and managing user authentication. passlib[bcrypt] is used for securely hashing passwords, while python-jose[cryptography] handles JSON Web Tokens (JWTs) for stateless authentication.

Python

# auth.py (a new file)
from passlib.context import CryptContext
from jose import jwt
from datetime import datetime, timedelta

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_delta: timedelta):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# main.py (continued)
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        return username
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="Could not validate credentials")

@app.get("/users/me")
def read_users_me(current_user: str = Depends(get_current_user)):
    return {"username": current_user}

4. Background Tasks ⏳

Use Case: Offloading long-running or resource-intensive tasks to a separate worker process. Celery is a robust and widely-used task queue for this purpose, keeping your API responsive.

Python

# tasks.py
from celery import Celery
import time

celery_app = Celery("tasks", broker="redis://localhost:6379/0")

@celery_app.task
def send_email_task(email_address: str):
    time.sleep(10)  # Simulate email sending
    return f"Email sent to {email_address}"

# main.py (continued)
@app.post("/send-email/")
def send_email(email_address: str):
    send_email_task.delay(email_address)
    return {"message": "Email sending initiated."}

5. Templating and Frontend 🖼️

Use Case: Rendering dynamic HTML pages for a full-stack application. Jinja2 is the industry standard templating engine for Python web frameworks. python-multipart is needed to parse HTML forms and file uploads.

Python

# main.py (continued)
from fastapi.templating import Jinja2Templates
from fastapi import Request
from fastapi.responses import HTMLResponse

templates = Jinja2Templates(directory="templates")

@app.get("/items/{id}", response_class=HTMLResponse)
async def read_item(request: Request, id: str):
    return templates.TemplateResponse(
        "item.html", {"request": request, "id": id}
    )

# templates/item.html
"""



    Item Page


    
Item ID: {{ id }}



"""

6. GraphQL 💡

Use Case: Building a GraphQL API. Strawberry and Ariadne are two popular libraries that integrate seamlessly with FastAPI to create a GraphQL endpoint. They use Python type hints to define your schema.

Python

# main.py (continued)
import strawberry
from strawberry.fastapi import GraphQLRouter
from typing import List

@strawberry.type
class User:
    name: str
    age: int

@strawberry.field
class Query:
    def all_users() -> List[User]:
        return [User(name="Alice", age=30), User(name="Bob", age=25)]

schema = strawberry.Schema(query=Query)
graphql_app = GraphQLRouter(schema)

app.include_router(graphql_app, prefix="/graphql")

7. Testing 🧪

Use Case: Writing tests for your API endpoints. While FastAPI includes a TestClient built on httpx, you'll typically use a testing framework like pytest to structure and run your tests.

Python

# test_main.py (a new file)
from fastapi.testclient import TestClient
from .main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_read_item():
    response = client.get("/items/foo")
    assert response.status_code == 200
    assert response.json() == {"item_id": "foo"}

Comments

Post a Comment

Popular posts from this blog

Fast API Built In Features

Fast API Built In Features Fast API Built In Features A look at the key components that make FastAPI so powerful. Core Framework (Starlette) This forms the fundamental structure of your application. Applications, Routing, Endpoints, Middleware, Responses, WebSockets, Exceptions: All of these are core functionalities inherited from Starlette. Use Cases: Routing: Define how your application responds to different URL paths (`@app.get("/items/{item_id}")`). Middleware: Perform actions on every request and response, such as adding custom headers, handling CORS, or logging request details. WebSockets: Build real-time applications like a live chat or a stock ticker that pushes data to the client as it becomes a...
Read more