FastAPI

1

tell me 20 features of FastAPI with code example

Here are 20 features of FastAPI along with code examples to demonstrate each feature:

1. Fast: FastAPI is one of the fastest Python web frameworks available.

   Copy

   from fastapi import FastAPI
   
   app = FastAPI()
   
   @app.get("/")
   async def read_root():
       return {"message": "Hello, FastAPI!"}

2. Path Parameters: Define dynamic paths with parameters.

   Copy

   @app.get("/items/{item_id}")
   async def read_item(item_id: int):
       return {"item_id": item_id}

3. Query Parameters: Support for query parameters.

   Copy

   @app.get("/items/")
   async def read_item(skip: int = 0, limit: int = 10):
       return {"skip": skip, "limit": limit}

4. Request Body: Accept request bodies with Pydantic models.

   Copy

   from pydantic import BaseModel
   
   class Item(BaseModel):
       name: str
       description: str = None
       price: float
   
   @app.post("/items/")
   async def create_item(item: Item):
       return item

5. Form Data: Accept form data in requests.

   Copy

   from fastapi import Form
   
   @app.post("/login/")
   async def login(username: str = Form(...), password: str = Form(...)):
       return {"username": username}

6. File Uploads: Handle file uploads.

   Copy

   from fastapi import File, UploadFile
   
   @app.post("/uploadfile/")
   async def create_upload_file(file: UploadFile = File(...)):
       return {"filename": file.filename}

7. Path Operations: Different HTTP methods (GET, POST, PUT, DELETE).

   Copy

   @app.put("/items/{item_id}")
   async def update_item(item_id: int, item: Item):
       return {"item_id": item_id, **item.dict()}

8. Response Model: Define response models.

   Copy

   class ItemResponse(BaseModel):
       name: str
       price: float
   
   @app.get("/items/{item_id}", response_model=ItemResponse)
   async def read_item(item_id: int):
       return {"name": "example", "price": 10.0}

9. Dependencies: Dependency injection system.

   Copy

   from fastapi import Depends
   
   def get_token(token: str):
       return token
   
   @app.get("/users/")
   async def read_users(token: str = Depends(get_token)):
       return {"token": token}

10. Security: OAuth2 with password and bearer token.

    Copy

    from fastapi.security import OAuth2PasswordBearer
    
    oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
    
    @app.get("/users/me")
    async def read_users_me(token: str = Depends(oauth2_scheme)):
        return {"token": token}

11. Background Tasks: Run background tasks.

    Copy

    from fastapi import BackgroundTasks
    
    def write_log(message: str):
        with open("log.txt", "a") as log:
            log.write(message)
    
    @app.post("/send-notification/{email}")
    async def send_notification(email: str, background_tasks: BackgroundTasks):
        background_tasks.add_task(write_log, f"Notification sent to {email}")
        return {"message": "Notification sent"}

12. WebSockets: Support for WebSocket.

    Copy

    from fastapi import WebSocket
    
    @app.websocket("/ws")
    async def websocket_endpoint(websocket: WebSocket):
        await websocket.accept()
        await websocket.send_text("Hello, WebSocket!")
        await websocket.close()

13. Middleware: Add middleware for handling requests and responses.

    Copy

    from fastapi.middleware.cors import CORSMiddleware
    
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

14. Events: Startup and shutdown events.

    Copy

    @app.on_event("startup")
    async def startup_event():
        print("Starting up...")
    
    @app.on_event("shutdown")
    async def shutdown_event():
        print("Shutting down...")

15. Templates: Render HTML templates.

    Copy

    from fastapi.templating import Jinja2Templates
    from fastapi import Request
    
    templates = Jinja2Templates(directory="templates")
    
    @app.get("/items/{id}")
    async def read_item(request: Request, id: str):
        return templates.TemplateResponse("item.html", {"request": request, "id": id})

16. Static Files: Serve static files.

    Copy

    from fastapi.staticfiles import StaticFiles
    
    app.mount("/static", StaticFiles(directory="static"), name="static")

17. Custom Responses: Return custom responses.

    Copy

    from fastapi.responses import JSONResponse
    
    @app.get("/custom_response")
    async def custom_response():
        return JSONResponse(content={"message": "Custom Response"})

18. Database Integration: Easily integrate with databases (example with SQLAlchemy).

    Copy

    from sqlalchemy import create_engine, Column, Integer, String, Sequence
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    
    SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
    engine = create_engine(SQLALCHEMY_DATABASE_URL)
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base = declarative_base()
    
    class User(Base):
        __tablename__ = "users"
        id = Column(Integer, Sequence("user_id_seq"), primary_key=True, index=True)
        name = Column(String, index=True)
    
    Base.metadata.create_all(bind=engine)
    
    @app.get("/users/{user_id}")
    async def read_user(user_id: int, db: Session = Depends(get_db)):
        user = db.query(User).filter(User.id == user_id).first()
        return user

19. Asynchronous Support: First-class support for async and await.

    Copy

    @app.get("/async")
    async def read_async():
        await some_async_function()
        return {"message": "Async function called"}

20. OpenAPI and JSON Schema: Automatically generated OpenAPI and JSON Schema docs.

    Copy

    # No code needed; FastAPI automatically generates this at /docs and /redoc

These features and examples demonstrate the versatility and powerful capabilities of FastAPI.

---

2

tell me another 20

Sure! Here are 20 more features of FastAPI along with code examples:

1. Custom Request Validation: Define custom validation logic for requests.

   Copy

   from fastapi import HTTPException, Request
   
   @app.post("/validate/")
   async def validate_request(request: Request):
       data = await request.json()
       if "key" not in data:
           raise HTTPException(status_code=400, detail="Key not found in request")
       return {"message": "Valid request"}

2. Custom Exceptions: Define custom exception handlers.

   Copy

   from fastapi.responses import JSONResponse
   
   class UnicornException(Exception):
       def __init__(self, name: str):
           self.name = name
   
   @app.exception_handler(UnicornException)
   async def unicorn_exception_handler(request: Request, exc: UnicornException):
       return JSONResponse(status_code=418, content={"message": f"Oops! {exc.name} did something."})
   
   @app.get("/unicorns/{name}")
   async def read_unicorn(name: str):
       if name == "yolo":
           raise UnicornException(name=name)
       return {"unicorn_name": name}

3. Custom Status Codes: Return custom status codes.

   Copy

   from fastapi import status
   
   @app.post("/status/", status_code=status.HTTP_201_CREATED)
   async def create_item():
       return {"message": "Item created"}

4. Custom Headers: Set custom headers in responses.

   Copy

   from fastapi import Response
   
   @app.get("/headers/")
   async def custom_headers(response: Response):
       response.headers["X-Custom-Header"] = "Custom value"
       return {"message": "Custom headers"}

5. Rate Limiting: Implement rate limiting (using third-party libraries like slowapi).

   Copy

   from slowapi import Limiter
   from slowapi.util import get_remote_address
   
   limiter = Limiter(key_func=get_remote_address)
   app.state.limiter = limiter
   
   @app.get("/limited/")
   @limiter.limit("5/minute")
   async def limited():
       return {"message": "This is a rate-limited endpoint"}

6. OAuth2 with JWT: OAuth2 authentication with JWT tokens.

   Copy

   from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
   from jose import JWTError, jwt
   
   oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
   
   @app.post("/token")
   async def login(form_data: OAuth2PasswordRequestForm = Depends()):
       # Token generation logic here
       return {"access_token": "token", "token_type": "bearer"}
   
   @app.get("/users/me")
   async def read_users_me(token: str = Depends(oauth2_scheme)):
       # Token verification logic here
       return {"token": token}

7. GraphQL: Integrate with GraphQL using graphene.

   Copy

   from fastapi import FastAPI
   from starlette.graphql import GraphQLApp
   import graphene
   
   class Query(graphene.ObjectType):
       hello = graphene.String(name=graphene.String(default_value="stranger"))
   
       def resolve_hello(self, info, name):
           return f"Hello {name}"
   
   app.add_route("/graphql", GraphQLApp(schema=graphene.Schema(query=Query)))

8. Event Handling: Handle application-wide events.

   Copy

   @app.on_event("startup")
   async def on_startup():
       # Startup event handling logic
       pass
   
   @app.on_event("shutdown")
   async def on_shutdown():
       # Shutdown event handling logic
       pass

9. CORS: Configure CORS (Cross-Origin Resource Sharing).

   Copy

   from fastapi.middleware.cors import CORSMiddleware
   
   app.add_middleware(
       CORSMiddleware,
       allow_origins=["*"],
       allow_credentials=True,
       allow_methods=["*"],
       allow_headers=["*"],
   )

10. HTTP Basic Auth: Implement basic HTTP authentication.

    Copy

    from fastapi.security import HTTPBasic, HTTPBasicCredentials
    
    security = HTTPBasic()
    
    @app.get("/users/me")
    async def read_current_user(credentials: HTTPBasicCredentials = Depends(security)):
        return {"username": credentials.username}

11. Internationalization (i18n): Support for multiple languages.

    Copy

    from fastapi_babel import Babel
    
    babel = Babel(app)
    
    @app.get("/greet/")
    async def greet():
        return babel.gettext("Hello, world!")

12. APIRouter: Use routers to organize your application.

    Copy

    from fastapi import APIRouter
    
    router = APIRouter()
    
    @router.get("/items/")
    async def read_items():
        return [{"item_id": "Foo"}, {"item_id": "Bar"}]
    
    app.include_router(router, prefix="/v1")

13. SQLAlchemy Integration: Use SQLAlchemy for database interactions.

    Copy

    from sqlalchemy import create_engine, Column, Integer, String
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy.orm import sessionmaker
    
    DATABASE_URL = "sqlite:///./test.db"
    
    engine = create_engine(DATABASE_URL)
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base = declarative_base()
    
    class User(Base):
        __tablename__ = "users"
        id = Column(Integer, primary_key=True, index=True)
        name = Column(String, index=True)
    
    Base.metadata.create_all(bind=engine)
    
    @app.get("/users/")
    async def read_users(db: Session = Depends(get_db)):
        users = db.query(User).all()
        return users

14. Alembic Integration: Database migrations with Alembic.

    Copy

    # Initialize Alembic for your project and use the command-line tools for migrations.

15. Dependency Overrides: Override dependencies for testing or specific use cases.

    Copy

    def get_test_db():
        return SessionLocal()
    
    app.dependency_overrides[get_db] = get_test_db

16. Custom Routers: Create custom routers for different modules.

    Copy

    from fastapi import APIRouter
    
    user_router = APIRouter()
    
    @user_router.get("/users/")
    async def read_users():
        return [{"username": "test"}]
    
    app.include_router(user_router, prefix="/api")

17. WebSocket Authentication: Handle authentication for WebSocket connections.

    Copy

    @app.websocket("/ws")
    async def websocket_endpoint(websocket: WebSocket, token: str):
        await websocket.accept()
        # Perform token verification
        await websocket.send_text("Authenticated WebSocket!")
        await websocket.close()

18. Custom URL Converters: Create custom URL converters for routes.

    Copy

    from fastapi import Path
    
    @app.get("/items/{item_id}")
    async def read_item(item_id: str = Path(..., regex="^[a-zA-Z0-9_-]+$")):
        return {"item_id": item_id}

19. CLI Commands: Create custom CLI commands.

    Copy

    import typer
    
    cli = typer.Typer()
    
    @cli.command()
    def create_user(username: str):
        print(f"Creating user: {username}")
    
    if __name__ == "__main__":
        cli()

20. Environment Variables: Use environment variables for configuration.

    Copy

    from pydantic import BaseSettings
    
    class Settings(BaseSettings):
        database_url: str
    
    settings = Settings()
    
    @app.get("/settings")
    async def get_settings():
        return {"database_url": settings.database_url}

These additional features and examples further illustrate the comprehensive capabilities of FastAPI, making it a versatile and powerful framework for building APIs.

---

3

tell me anoter 20

Of course! Here are 20 more features of FastAPI along with code examples:

1. Middleware for Logging: Implement logging with middleware.

   Copy

   import time
   from fastapi.middleware import Middleware
   
   @app.middleware("http")
   async def log_requests(request: Request, call_next):
       start_time = time.time()
       response = await call_next(request)
       process_time = time.time() - start_time
       log_message = f"{request.method} {request.url} {response.status_code} {process_time}"
       print(log_message)
       return response

2. Custom Response Class: Create and use custom response classes.

   Copy

   from fastapi.responses import PlainTextResponse
   
   @app.get("/custom_response", response_class=PlainTextResponse)
   async def custom_response():
       return "This is a plain text response"

3. Cookies: Read and set cookies in responses.

   Copy

   from fastapi import Cookie
   
   @app.get("/get-cookie/")
   async def read_cookie(cookie_id: str = Cookie(None)):
       return {"cookie_id": cookie_id}
   
   @app.post("/set-cookie/")
   async def set_cookie(response: Response):
       response.set_cookie(key="cookie_id", value="cookie_value")
       return {"message": "Cookie set"}

4. Custom Exception Messages: Customize exception messages globally.

   Copy

   from fastapi.responses import JSONResponse
   
   @app.exception_handler(HTTPException)
   async def custom_http_exception_handler(request: Request, exc: HTTPException):
       return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
   
   @app.get("/error/")
   async def raise_error():
       raise HTTPException(status_code=404, detail="Custom Not Found")

5. Background Task Cleanup: Use background tasks for cleanup.

   Copy

   from fastapi import BackgroundTasks
   
   def cleanup_task():
       print("Cleanup task running")
   
   @app.get("/cleanup/")
   async def schedule_cleanup(background_tasks: BackgroundTasks):
       background_tasks.add_task(cleanup_task)
       return {"message": "Cleanup scheduled"}

6. Streaming Responses: Stream large responses.

   Copy

   from fastapi.responses import StreamingResponse
   
   async def generate_numbers():
       for i in range(10):
           yield str(i)
           await asyncio.sleep(1)
   
   @app.get("/stream/")
   async def stream_response():
       return StreamingResponse(generate_numbers())

7. Mount Sub-applications: Mount other ASGI applications.

   Copy

   from starlette.applications import Starlette
   
   sub_app = Starlette()
   
   @sub_app.route("/sub")
   async def sub_app_route(request):
       return JSONResponse({"message": "Hello from sub app"})
   
   app.mount("/subapp", sub_app)

8. HTTP Redirects: Return HTTP redirects.

   Copy

   from fastapi.responses import RedirectResponse
   
   @app.get("/redirect/")
   async def redirect():
       return RedirectResponse(url="/new-url")
   
   @app.get("/new-url/")
   async def new_url():
       return {"message": "You have been redirected"}

9. File Responses: Serve files as responses.

   Copy

   from fastapi.responses import FileResponse
   
   @app.get("/download/")
   async def download_file():
       return FileResponse("path/to/file.txt", filename="file.txt")

10. GraphQL Subscriptions: Implement GraphQL subscriptions.

    Copy

    # Using the Ariadne library for subscriptions.
    from ariadne import gql, make_executable_schema
    from ariadne.asgi import GraphQL
    
    type_defs = gql("""
    type Query {
        hello: String!
    }
    type Subscription {
        count: Int!
    }
    """)
    
    async def resolve_hello(_, info):
        return "Hello!"
    
    async def resolve_count(_, info):
        for i in range(10):
            yield {"count": i}
            await asyncio.sleep(1)
    
    resolvers = {
        "Query": {"hello": resolve_hello},
        "Subscription": {"count": resolve_count}
    }
    
    schema = make_executable_schema(type_defs, resolvers)
    app.mount("/graphql", GraphQL(schema, debug=True))

11. Versioning: API versioning with path parameters.

    Copy

    @app.get("/v1/items/")
    async def read_v1_items():
        return {"message": "API version 1"}
    
    @app.get("/v2/items/")
    async def read_v2_items():
        return {"message": "API version 2"}

12. OAuth2 Password Flow: Implement OAuth2 password flow with JWT.

    Copy

    from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
    
    oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
    
    @app.post("/token")
    async def generate_token(form_data: OAuth2PasswordRequestForm = Depends()):
        return {"access_token": "fake-jwt-token", "token_type": "bearer"}
    
    @app.get("/users/me")
    async def read_users_me(token: str = Depends(oauth2_scheme)):
        return {"token": token}

13. Dependency Injection: Advanced use of dependency injection.

    Copy

    def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
        return {"q": q, "skip": skip, "limit": limit}
    
    @app.get("/items/")
    async def read_items(commons: dict = Depends(common_parameters)):
        return commons

14. Data Validation with Pydantic: Use Pydantic for advanced data validation.

    Copy

    from pydantic import BaseModel, Field
    
    class Item(BaseModel):
        name: str = Field(..., max_length=50)
        price: float = Field(..., gt=0, lt=1000)
    
    @app.post("/items/")
    async def create_item(item: Item):
        return item

15. Custom Validation Error Handler: Handle validation errors globally.

    Copy

    from fastapi.exceptions import RequestValidationError
    from fastapi.responses import JSONResponse
    
    @app.exception_handler(RequestValidationError)
    async def validation_exception_handler(request: Request, exc: RequestValidationError):
        return JSONResponse(status_code=400, content={"detail": exc.errors()})

16. API Key Security: Implement API key security.

    Copy

    from fastapi.security.api_key import APIKeyHeader
    
    api_key_header = APIKeyHeader(name="X-API-Key")
    
    @app.get("/protected/")
    async def protected_route(api_key: str = Depends(api_key_header)):
        if api_key != "expected-api-key":
            raise HTTPException(status_code=403, detail="Invalid API key")
        return {"message": "Access granted"}

17. GraphQL with Ariadne: Use Ariadne for GraphQL endpoints.

    Copy

    from ariadne import gql, QueryType, make_executable_schema
    from ariadne.asgi import GraphQL
    
    type_defs = gql("""
        type Query {
            hello: String!
        }
    """)
    
    query = QueryType()
    
    @query.field("hello")
    def resolve_hello(_, info):
        return "Hello, world!"
    
    schema = make_executable_schema(type_defs, query)
    app.mount("/graphql", GraphQL(schema, debug=True))

18. AsyncSQLAlchemy: Use async support with SQLAlchemy.

    Copy

    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.orm import sessionmaker
    
    DATABASE_URL = "sqlite+aiosqlite:///./test.db"
    engine = create_async_engine(DATABASE_URL, echo=True)
    AsyncSessionLocal = sessionmaker(
        bind=engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )
    
    async def get_async_db():
        async with AsyncSessionLocal() as session:
            yield session

19. Testing with TestClient: Write tests using FastAPI's TestClient.

    Copy

    from fastapi.testclient import TestClient
    
    client = TestClient(app)
    
    def test_read_main():
        response = client.get("/")
        assert response.status_code == 200
        assert response.json() == {"message": "Hello, FastAPI!"}

20. Handle Multiple Applications: Serve multiple FastAPI applications.

    Copy

    from fastapi import FastAPI
    
    app1 = FastAPI()
    app2 = FastAPI()
    
    @app1.get("/")
    async def read_root1():
        return {"message": "Hello from app1"}
    
    @app2.get("/")
    async def read_root2():
        return {"message": "Hello from app2"}
    
    from fastapi.middleware.wsgi import WSGIMiddleware
    app = FastAPI()
    app.mount("/app1", WSGIMiddleware(app1))
    app.mount("/app2", WSGIMiddleware(app2))

21. Role-based Access Control (RBAC): Implement RBAC.

    Copy

    from fastapi import Security
    from fastapi.security import OAuth2PasswordBearer, SecurityScopes
    
    oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", scopes={"admin": "Admin access", "user": "User access"})
    
    def get_current_user(security_scopes: SecurityScopes, token

: str = Depends(oauth2_scheme)): if security_scopes.scopes and "admin" not in security_scopes.scopes: raise HTTPException(status_code=403, detail="Not enough permissions") return {"username": "admin"}

Copy

@app.get("/admin/")
async def read_admin_data(current_user: dict = Security(get_current_user, scopes=["admin"])):
    return {"message": "Admin data"}
```

62. Response Compression: Enable Gzip compression for responses. ```python from fastapi.middleware.gzip import GZipMiddleware

Copy

app.add_middleware(GZipMiddleware, minimum_size=1000)
```

63. Locale Detection: Automatically detect and handle user locales. ```python from fastapi_babel import Babel

Copy

babel = Babel(app)

@babel.localeselector
def get_locale():
    return request.headers.get("Accept-Language")

@app.get("/greet/")
async def greet():
    return babel.gettext("Hello, world!")
```

64. Token-Based Rate Limiting: Implement token-based rate limiting. ```python from slowapi import Limiter from slowapi.util import get_remote_address

Copy

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/limited/")
@limiter.limit("10/minute")
async def limited():
    return {"message": "This is a rate-limited endpoint"}
```

65. WebSocket Broadcast: Broadcast messages to multiple WebSocket clients. ```python from fastapi import WebSocket, WebSocketDisconnect from typing import List

Copy

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
```

66. Dependency Override in Tests: Override dependencies in tests. ```python from fastapi.testclient import TestClient

Copy

app.dependency_overrides[get_db] = get_test_db
client = TestClient(app)

def test_override():
    response = client.get("/items/")
    assert response.status_code == 200
    assert response.json() == {"items": []}
```

67. Request Hooks: Implement hooks to process requests before they reach the route. python @app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time response.headers["X-Process-Time"] = str(process_time) return response

1. FastAPI with Celery: Integrate FastAPI with Celery for task queues.

   Copy

   from celery import Celery
   
   celery_app = Celery(__name__, broker="redis://localhost:6379/0")
   
   @celery_app.task
   def add(x, y):
       return x + y
   
   @app.post("/tasks/")
   async def create_task(x: int, y: int):
       task = add.apply_async((x, y))
       return {"task_id": task.id}

2. SQLAlchemy with Alembic: Integrate SQLAlchemy with Alembic for migrations.

   Copy

   # Initialize Alembic in your project and use the command-line tools for database migrations.

3. Admin Interface: Integrate with an admin interface (e.g., fastapi-admin).

   Copy

   from fastapi_admin.factory import app as admin_app
   
   @app.on_event("startup")
   async def startup():
       await admin_app.init(
           app, 
           database_url="sqlite:///./test.db", 
           secret_key="SECRET_KEY"
       )

These features, along with the previously mentioned ones, illustrate the extensive functionality and flexibility that FastAPI offers for building modern web APIs and applications.

---

4

give me another 20

Sure! Here are another 20 features of FastAPI along with code examples:

1. Custom Dependency Classes: Use custom classes as dependencies.

   Copy

   class CommonQueryParams:
       def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
           self.q = q
           self.skip = skip
           self.limit = limit
   
   @app.get("/items/")
   async def read_items(commons: CommonQueryParams = Depends()):
       return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

2. SQLAlchemy with FastAPI: Integrate SQLAlchemy ORM with FastAPI.

   Copy

   from sqlalchemy.orm import declarative_base, sessionmaker
   from sqlalchemy import create_engine
   
   SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
   engine = create_engine(SQLALCHEMY_DATABASE_URL)
   SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
   Base = declarative_base()
   
   @app.middleware("http")
   async def db_session_middleware(request: Request, call_next):
       response = Response("Internal server error", status_code=500)
       try:
           request.state.db = SessionLocal()
           response = await call_next(request)
       finally:
           request.state.db.close()
       return response

3. Testing Dependencies: Override dependencies in tests.

   Copy

   from fastapi.testclient import TestClient
   
   def override_dependency():
       return {"override": "dependency"}
   
   app.dependency_overrides[dependency] = override_dependency
   
   client = TestClient(app)
   
   def test_override_dependency():
       response = client.get("/dependency/")
       assert response.json() == {"override": "dependency"}

4. Peewee ORM: Use Peewee ORM with FastAPI.

   Copy

   from peewee import SqliteDatabase, Model, CharField
   
   db = SqliteDatabase('test.db')
   
   class User(Model):
       name = CharField()
   
       class Meta:
           database = db
   
   @app.on_event("startup")
   def startup():
       db.connect()
       db.create_tables([User])
   
   @app.on_event("shutdown")
   def shutdown():
       db.close()
   
   @app.get("/users/")
   async def read_users():
       users = User.select()
       return [{"name": user.name} for user in users]

5. CORS Configuration: Configure CORS for specific routes.

   Copy

   from fastapi.middleware.cors import CORSMiddleware
   
   app.add_middleware(
       CORSMiddleware,
       allow_origins=["https://example.com"],
       allow_credentials=True,
       allow_methods=["GET"],
       allow_headers=["*"],
   )

6. Custom Field Validation: Use Pydantic for custom field validation.

   Copy

   from pydantic import BaseModel, validator
   
   class Item(BaseModel):
       name: str
       price: float
   
       @validator('price')
       def price_must_be_positive(cls, v):
           if v <= 0:
               raise ValueError('Price must be positive')
           return v
   
   @app.post("/items/")
   async def create_item(item: Item):
       return item

7. Event-driven Architecture: Use event handlers for different stages.

   Copy

   @app.on_event("startup")
   async def startup_event():
       print("Application startup")
   
   @app.on_event("shutdown")
   async def shutdown_event():
       print("Application shutdown")

8. WebSocket Middleware: Implement middleware for WebSocket connections.

   Copy

   from starlette.middleware.base import BaseHTTPMiddleware
   
   class WebSocketMiddleware(BaseHTTPMiddleware):
       async def dispatch(self, request, call_next):
           if request.url.path == "/ws":
               # Custom WebSocket handling
               pass
           return await call_next(request)
   
   app.add_middleware(WebSocketMiddleware)

9. Dynamic URL Path: Define dynamic URL paths with custom path converters.

   Copy

   from fastapi import Path
   
   @app.get("/items/{item_id}")
   async def read_item(item_id: str = Path(..., regex="^[a-zA-Z0-9_-]+$")):
       return {"item_id": item_id}

10. Conditional Response: Return different responses based on conditions.

    Copy

    from fastapi.responses import JSONResponse
    
    @app.get("/conditional/")
    async def conditional_response(condition: bool):
        if condition:
            return JSONResponse(content={"message": "Condition is true"})
        else:
            return JSONResponse(content={"message": "Condition is false"})

11. Dynamic Middleware: Apply middleware dynamically.

    Copy

    from starlette.middleware.base import BaseHTTPMiddleware
    
    class CustomMiddleware(BaseHTTPMiddleware):
        async def dispatch(self, request, call_next):
            response = await call_next(request)
            response.headers['X-Custom-Header'] = 'value'
            return response
    
    app.add_middleware(CustomMiddleware)

12. Global Exception Handling: Handle exceptions globally.

    Copy

    @app.exception_handler(Exception)
    async def global_exception_handler(request: Request, exc: Exception):
        return JSONResponse(status_code=500, content={"detail": str(exc)})
    
    @app.get("/cause_exception/")
    async def cause_exception():
        raise Exception("This is an exception")

13. Custom 404 Page: Customize the 404 error page.

    Copy

    from fastapi.responses import HTMLResponse
    
    @app.exception_handler(404)
    async def custom_404_handler(request: Request, exc):
        return HTMLResponse(content="<h1>Page not found</h1>", status_code=404)

14. Basic Authentication: Implement HTTP Basic Authentication.

    Copy

    from fastapi.security import HTTPBasic, HTTPBasicCredentials
    
    security = HTTPBasic()
    
    @app.get("/secure/")
    async def secure(credentials: HTTPBasicCredentials = Depends(security)):
        return {"username": credentials.username}

15. Advanced Dependency Injection: Use multiple levels of dependencies.

    Copy

    def dependency1():
        return "dependency1"
    
    def dependency2(dep1=Depends(dependency1)):
        return f"dependency2 uses {dep1}"
    
    @app.get("/complex/")
    async def complex_dependency(dep2=Depends(dependency2)):
        return {"dep2": dep2}

16. Database Connection Pooling: Implement connection pooling.

    Copy

    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.orm import sessionmaker
    
    SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
    engine = create_async_engine(SQLALCHEMY_DATABASE_URL, pool_size=5, max_overflow=10)
    AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
    
    @app.middleware("http")
    async def db_session_middleware(request: Request, call_next):
        response = Response("Internal server error", status_code=500)
        try:
            request.state.db = AsyncSessionLocal()
            response = await call_next(request)
        finally:
            await request.state.db.close()
        return response

17. Custom Dependencies with State: Use FastAPI state to share dependencies.

    Copy

    app.state.custom_value = "value"
    
    def get_custom_value():
        return app.state.custom_value
    
    @app.get("/state/")
    async def read_state(custom_value: str = Depends(get_custom_value)):
        return {"custom_value": custom_value}

18. Dependency Injection with Context Managers: Use context managers as dependencies.

    Copy

    from contextlib import contextmanager
    
    @contextmanager
    def context_dependency():
        yield "context value"
    
    @app.get("/context/")
    async def context_example(value: str = Depends(context_dependency)):
        return {"value": value}

19. GraphQL with Strawberry: Use Strawberry for GraphQL API.

    Copy

    import strawberry
    from strawberry.asgi import GraphQL
    
    @strawberry.type
    class Query:
        hello: str = "Hello, world!"
    
    schema = strawberry.Schema(Query)
    app.mount("/graphql", GraphQL(schema))

20. HTML Forms Handling: Handle HTML forms with FastAPI.

    Copy

    from fastapi import Form
    
    @app.post("/login/")
    async def login(username: str = Form(...), password: str = Form(...)):
        return {"username": username, "password": password}

21. FastAPI with Django ORM: Use FastAPI with Django ORM.

    Copy

    import django
    from django.conf import settings
    
    settings.configure(
        INSTALLED_APPS=[
            'your_django_app',
        ],
        DATABASES={
            'default': {
                'ENGINE': 'django.db.backends.sqlite3',
                'NAME': 'mydatabase',
            }
        }
    )
    django.setup()
    
    @app.get("/django/")
    async def read_django():
        from your_django_app.models import YourModel
        items = YourModel.objects.all()
        return [{"name": item.name} for item in items]

22. SQLAlchemy with PostgreSQL: Use SQLAlchemy with PostgreSQL.

    Copy

    from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
    from sqlalchemy.orm import sessionmaker
    
    
    
    SQLALCHEMY_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
    engine = create_async_engine(SQLALCHEMY_DATABASE_URL)
    AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
    
    @app.get("/postgres/")
    async def read_postgres():
        async with AsyncSessionLocal() as session:
            result = await session.execute("SELECT 1")
            return {"result": result.scalar()}

23. Data Encryption: Encrypt and decrypt data in FastAPI.

    Copy

    from cryptography.fernet import Fernet
    
    key = Fernet.generate_key()
    cipher_suite = Fernet(key)
    
    @app.post("/encrypt/")
    async def encrypt(data: str):
        encrypted_data = cipher_suite.encrypt(data.encode())
        return {"encrypted_data": encrypted_data}
    
    @app.post("/decrypt/")
    async def decrypt(encrypted_data: str):
        decrypted_data = cipher_suite.decrypt(encrypted_data.encode())
        return {"decrypted_data": decrypted_data.decode()}

24. Rate Limiting with Redis: Use Redis for rate limiting.

    Copy

    from fastapi_limiter import FastAPILimiter
    
    @app.on_event("startup")
    async def startup():
        redis = await aioredis.create_redis_pool("redis://localhost")
        await FastAPILimiter.init(redis)
    
    @app.get("/limited/")
    @limiter.limit("5/minute")
    async def limited():
        return {"message": "This is a rate-limited endpoint"}

25. WebSocket Authentication with JWT: Authenticate WebSocket connections with JWT.

    Copy

    @app.websocket("/ws")
    async def websocket_endpoint(websocket: WebSocket, token: str):
        await websocket.accept()
        # Verify token
        await websocket.send_text("Authenticated WebSocket")
        await websocket.close()

26. Throttling: Implement request throttling.

    Copy

    from fastapi_limiter.depends import RateLimiter
    
    @app.get("/throttle/")
    @RateLimiter(times=10, seconds=60)
    async def throttle():
        return {"message": "Request throttled"}

27. HTML Template Rendering: Render HTML templates with Jinja2.

    Copy

    from fastapi.templating import Jinja2Templates
    
    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})

28. Rate Limiting with FastAPI-Limiter: Use fastapi-limiter for rate limiting.

    Copy

    from fastapi_limiter import FastAPILimiter
    
    @app.on_event("startup")
    async def on_startup():
        redis = await aioredis.create_redis_pool("redis://localhost")
        FastAPILimiter.init(redis)
    
    @app.get("/limited/", dependencies=[Depends(RateLimiter(times=10, seconds=60))])
    async def limited():
        return {"message": "This endpoint is rate-limited"}

29. GraphQL with Graphene: Use Graphene for GraphQL APIs.

    Copy

    import graphene
    from starlette.graphql import GraphQLApp
    
    class Query(graphene.ObjectType):
        hello = graphene.String(name=graphene.String(default_value="stranger"))
    
        def resolve_hello(self, info, name):
            return f"Hello {name}"
    
    app.add_route("/graphql", GraphQLApp(schema=graphene.Schema(query=Query)))

30. Event Notifications: Implement event notifications. ```python from pydantic import BaseModel

Copy

class Event(BaseModel):

Copy

    event_type: str
    payload: dict

@app.post("/event/")
async def create_event(event: Event):
    # Process event
    return {"message": "Event processed"}
```

These features and examples illustrate the extensive and diverse capabilities of FastAPI for creating robust and scalable APIs and applications.

---

5

I apologize for the previous issues. Here are the 20 features with properly formatted code examples:

101. Automatic Documentation Generation

FastAPI automatically generates interactive API documentation at /docs (Swagger UI) and /redoc (ReDoc).

Copy

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, FastAPI!"}
# Access interactive documentation at /docs and /redoc

102. Dependency Injection for Classes

Use class-based dependencies.

Copy

class Database:
    def __init__(self, url: str):
        self.url = url

def get_db():
    return Database(url="sqlite:///test.db")

@app.get("/db/")
async def read_db(db: Database = Depends(get_db)):
    return {"db_url": db.url}

103. GraphQL Subscriptions

Implement GraphQL subscriptions with WebSockets.

Copy

import graphene
from starlette.graphql import GraphQLApp
from graphql.execution.executors.asyncio import AsyncioExecutor

class Query(graphene.ObjectType):
    hello = graphene.String(name=graphene.String(default_value="stranger"))

    def resolve_hello(self, info, name):
        return f"Hello {name}"

app.add_route("/graphql", GraphQLApp(schema=graphene.Schema(query=Query), executor_class=AsyncioExecutor))

104. Redis Cache Integration

Use Redis for caching.

Copy

import aioredis

@app.on_event("startup")
async def startup():
    app.state.redis = await aioredis.create_redis_pool("redis://localhost")

@app.on_event("shutdown")
async def shutdown():
    app.state.redis.close()
    await app.state.redis.wait_closed()

@app.get("/cache/")
async def read_cache(key: str):
    value = await app.state.redis.get(key)
    return {"key": key, "value": value}

105. Dependency Overriding

Override dependencies for testing.

Copy

from fastapi.testclient import TestClient

def get_test_db():
    return Database(url="sqlite:///test.db")

app.dependency_overrides[get_db] = get_test_db

client = TestClient(app)

def test_read_db():
    response = client.get("/db/")
    assert response.json() == {"db_url": "sqlite:///test.db"}

106. Pydantic Settings Management

Use Pydantic for settings management.

Copy

from pydantic import BaseSettings

class Settings(BaseSettings):
    database_url: str

settings = Settings(database_url="sqlite:///test.db")

@app.get("/settings/")
async def read_settings():
    return {"database_url": settings.database_url}

107. Database Transactions

Use transactions with databases.

Copy

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)

@app.post("/transaction/")
async def create_transaction():
    async with AsyncSessionLocal() as session:
        async with session.begin():
            # Perform database operations
            pass
    return {"message": "Transaction completed"}

108. OpenAPI Customization

Customize the OpenAPI schema.

Copy

from fastapi.openapi.utils import get_openapi

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="Custom API",
        version="1.0.0",
        description="This is a custom OpenAPI schema",
        routes=app.routes,
    )
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

109. Async Task Queues with RQ

Use RQ for background tasks.

Copy

import redis
from rq import Queue
from time import sleep

r = redis.Redis()
q = Queue(connection=r)

def background_task(n):
    sleep(n)
    return f"Task completed in {n} seconds"

@app.post("/task/")
async def create_task(seconds: int):
    job = q.enqueue(background_task, seconds)
    return {"job_id": job.get_id()}

110. File Streaming

Stream files to the client.

Copy

from fastapi.responses import StreamingResponse

@app.get("/stream-file/")
async def stream_file():
    def iterfile():
        with open("largefile.txt", mode="rb") as file_like:
            yield from file_like

    return StreamingResponse(iterfile(), media_type="application/octet-stream")

111. Rate Limiting with IP Address

Implement rate limiting based on IP addresses.

Copy

from fastapi_limiter import FastAPILimiter
from fastapi_limiter.depends import RateLimiter

@app.on_event("startup")
async def startup():
    redis = await aioredis.create_redis_pool("redis://localhost")
    await FastAPILimiter.init(redis)

@app.get("/limited/")
@RateLimiter(times=5, seconds=60)
async def limited_endpoint():
    return {"message": "Rate limited endpoint"}

112. Streaming JSON Responses

Stream JSON responses to the client.

Copy

from fastapi.responses import StreamingResponse
import json

@app.get("/stream-json/")
async def stream_json():
    async def iter_json():
        for i in range(10):
            yield json.dumps({"number": i})
            await asyncio.sleep(1)

    return StreamingResponse(iter_json(), media_type="application/json")

113. Session Management

Use sessions for user management.

Copy

from fastapi import FastAPI, Request, Depends
from starlette.middleware.sessions import SessionMiddleware

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key="your-secret-key")

@app.get("/set-session/")
async def set_session(request: Request):
    request.session["user"] = "test_user"
    return {"message": "Session set"}

@app.get("/get-session/")
async def get_session(request: Request):
    user = request.session.get("user")
    return {"user": user}

114. WebSocket Chat

Implement a WebSocket chat server.

Copy

from fastapi import WebSocket
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def chat_websocket(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

115. Advanced Dependency Injection

Use advanced dependency injection with classes and methods.

Copy

class CommonQueryParams:
    def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

116. Request Validation

Custom request validation with Pydantic.

Copy

from pydantic import BaseModel, validator

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

    @validator('price')
    def price_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('Price must be positive')
        return v

@app.post("/items/")
async def create_item(item: Item):
    return item

117. OAuth2 with JWT

Implement OAuth2 authentication with JWT tokens.

Copy

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict):
    return jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    access_token = create_access_token(data={"sub": form_data.username})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(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 authentication credentials")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid authentication credentials")
    return {"username": username}

118. Handling Background Tasks

Run background tasks in FastAPI.

Copy

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", "a") as log:
        log.write(message)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message

": "Notification sent"}

119. WebSocket Authentication

Handle authentication for WebSocket connections.

Copy

from fastapi import WebSocket, Depends
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, "your_secret_key", algorithms=["HS256"])
        user = payload.get("sub")
        if user is None:
            raise HTTPException(status_code=401, detail="Invalid authentication credentials")
        return user
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid authentication credentials")

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(oauth2_scheme)):
    user = await get_current_user(token)
    await websocket.accept()
    await websocket.send_text(f"Hello {user}!")
    await websocket.close()

120. SQLAlchemy Integration with FastAPI

Use SQLAlchemy ORM with FastAPI.

Copy

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response

121. Role-Based Access Control (RBAC)

Implement RBAC with FastAPI.

Copy

from fastapi import Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", scopes={"admin": "Admin access", "user": "User access"})

def get_current_user(security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)):
    if security_scopes.scopes and "admin" not in security_scopes.scopes:
        raise HTTPException(status_code=403, detail="Not enough permissions")
    return {"username": "admin"}

@app.get("/admin/")
async def read_admin_data(current_user: dict = Security(get_current_user, scopes=["admin"])):
    return {"message": "Admin data"}

122. Custom Middleware

Create and use custom middleware in FastAPI.

Copy

from starlette.middleware.base import BaseHTTPMiddleware

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)
        response.headers['X-Custom-Header'] = 'value'
        return response

app.add_middleware(CustomMiddleware)

123. Request Validation with Dependencies

Validate requests with custom dependencies.

Copy

from fastapi import HTTPException

def verify_token(token: str):
    if token != "my-secret-token":
        raise HTTPException(status_code=400, detail="Invalid token")

@app.post("/verify-token/")
async def verify_token_endpoint(token: str = Depends(verify_token)):
    return {"message": "Token is valid"}

124. Dependency Injection for Multiple Levels

Use dependencies that depend on other dependencies.

Copy

def dependency_a():
    return "a"

def dependency_b(a: str = Depends(dependency_a)):
    return f"b uses {a}"

@app.get("/multiple-levels/")
async def read_multiple_levels(b: str = Depends(dependency_b)):
    return {"b": b}

125. Rate Limiting with FastAPI-Limiter

Implement rate limiting using the fastapi-limiter library.

Copy

from fastapi_limiter import FastAPILimiter

@app.on_event("startup")
async def startup():
    redis = await aioredis.create_redis_pool("redis://localhost")
    await FastAPILimiter.init(redis)

@app.get("/limited/")
@RateLimiter(times=5, seconds=60)
async def limited_endpoint():
    return {"message": "Rate limited endpoint"}

These additional features and examples further illustrate the flexibility and power of FastAPI in building modern web APIs and applications.

---

6

I have 4 routes in FastAPI. Out of 4, 3 of them having tenant_id like below

Copy

@app.get("/{tenant_id}/log-test")
async def log(request: Request):

I want to keep middleware where I would prepend tenant_id without specifically mention in each route.

To apply a middleware that automatically handles the tenant_id in FastAPI, you can create a middleware component that extracts the tenant_id from the path of incoming requests and then makes it conveniently available throughout your application, perhaps by attaching it to the request state.

Here's how you can define such middleware:

1. Create a Middleware: This middleware will capture requests, extract the tenant_id from the path if present, and then store it in the request state for use in your endpoint functions.

2. Use Starlette's Request Object: Since FastAPI is built on Starlette, you can utilize the request object's state to pass around the tenant_id.

Here's an example of how you might implement this:

Copy

from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import Response
import re

app = FastAPI()

class TenantMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        path = request.url.path
        match = re.search(r'^/(\d+)/', path)  # Assuming tenant_id is a digit group
        if match:
            tenant_id = match.group(1)
            request.state.tenant_id = tenant_id  # Store tenant_id in request state
        else:
            request.state.tenant_id = None  # Default or handle error

        response = await call_next(request)
        return response

# Add the middleware to the app
app.add_middleware(TenantMiddleware)

# Example endpoint using tenant_id from request state
@app.get("/{tenant_id}/log-test")
async def log_test(request: Request):
    tenant_id = getattr(request.state, 'tenant_id', None)
    if tenant_id is None:
        raise HTTPException(status_code=404, detail="Tenant ID not found")
    return {"message": "Logging for tenant", "tenant_id": tenant_id}

Explanation

• Middleware: TenantMiddleware checks each request's URL path for a tenant_id pattern. If found, it stores this tenant_id in the request.state, which is accessible throughout the lifecycle of the request.

• Regex Usage: The regex r'^/(\d+)/' is used to extract the first group of digits in the path, which is assumed to be the tenant_id. You can adjust the regex to match the specific format of your tenant_id.

• Endpoint: In the endpoint function, tenant_id is retrieved from request.state. If it's not found (i.e., None), an HTTP exception is raised. This centralizes the handling of tenant_id and avoids repeating the extraction logic in each endpoint.

By using this middleware, you simplify your route handlers by removing the need to manually extract the tenant_id from the path, ensuring that the ID handling logic is consistent and easily maintainable across your application.

---

7

Dependency Injection for Multiple Levels

how to use this in FastAPI in realtime

Dependency Injection (DI) in FastAPI is a powerful feature that allows you to decouple components and ensure that dependencies are managed automatically by the framework. This makes your code cleaner, easier to test, and more modular. FastAPI uses a system where dependencies are "injected" into your path operation functions as arguments.

Here’s a step-by-step guide to using dependency injection in FastAPI, especially focusing on multiple levels of dependencies:

1. Basic Dependency

A dependency in FastAPI can be any callable that returns a value. These are often used to provide database connections, configurations, or custom logic that can be reused across multiple endpoints.

Copy

from fastapi import FastAPI, Depends

app = FastAPI()

def get_query(token: str):
    return {"token": token}

@app.get("/items/")
async def read_items(query=Depends(get_query)):
    return {"query": query}

2. Multiple Level Dependencies

Dependencies can also depend on other dependencies, creating a chain or a tree of dependencies. This is useful when a particular service or component requires another service.

Copy

def get_token(header: str = Header(...)):
    return header.split(" ")[1]  # Assuming Bearer token

def get_query(token: str = Depends(get_token)):
    # Imagine this fetches and validates token then retrieves a query
    return {"token": token}

@app.get("/items/")
async def read_items(query=Depends(get_query)):
    return {"query": query}

3. Sub-dependencies

Here’s how you might define a sub-dependency where one dependency function relies on another. This allows you to build a complex but manageable dependency injection system.

Copy

from fastapi import Header

def get_database():
    db = "Connected to database"
    try:
        yield db
    finally:
        db = "Disconnected"  # Simulate disconnect

def get_user(db=Depends(get_database)):
    return {"db": db, "user_id": "user123"}

@app.get("/users/")
async def get_user_data(user=Depends(get_user)):
    return {"user": user}

4. Using Classes for Dependencies

You can also use classes for organizing dependencies. This is useful when your dependencies have state or configuration.

Copy

class DBConnection:
    def __init__(self, hostname: str):
        self.hostname = hostname

    def connect(self):
        return f"Connection to {self.hostname}"

def get_db_connection():
    db_connection = DBConnection(hostname="localhost")
    return db_connection.connect()

@app.get("/data/")
async def read_data(connection=Depends(get_db_connection)):
    return {"connection": connection}

5. Advanced Usage with Scopes

FastAPI allows you to specify how often a dependency should be reused. This can be configured using use_cache in the Depends function.

Copy

from fastapi import Depends, HTTPException, status

def get_token():
    token = "securetoken"
    return token

def get_admin_user(token=Depends(get_token, use_cache=False)):
    if token != "admin":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
    return {"token": token, "role": "admin"}

@app.get("/admin/")
async def get_admin_endpoint(user=Depends(get_admin_user)):
    return {"user": user}

Real-Time Considerations

When considering real-time applications or features:

• Cache Control: Managing the use_cache parameter carefully ensures that dependencies are re-evaluated as needed, which is critical in high-concurrency or rapidly-changing data environments.

• Async Dependencies: If your dependency involves IO operations (like database or network calls), it should ideally be asynchronous to avoid blocking the server loop.

• Error Handling: Dependencies can also perform error handling or security checks, which can preemptively return errors before reaching the actual business logic of your endpoints.

This approach to dependency injection keeps your FastAPI application modular, testable, and clean, enabling you to manage complex dependencies efficiently in real-time operations.

---

8

Give me some real time sample for "Error Handling" in FastAPI

Error handling in FastAPI is critical for managing exceptions and providing meaningful error responses to clients. FastAPI offers several built-in tools to streamline error handling, including exception handlers and dependency injections that can pre-validate data or manage security concerns. Here, I'll show you how to effectively use FastAPI’s features to handle errors in a practical, real-world application context.

1. Using HTTPException for Simple Error Handling

One of the simplest ways to handle errors in FastAPI is by raising an HTTPException when something goes wrong.

Copy

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id != 42:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

2. Custom Exception Handlers

You can define custom exception handlers to catch specific exceptions and return custom responses. This is particularly useful for catching broad exceptions that might be raised by libraries or your own code.

Copy

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": f"Oops! {exc.detail} did not work."},
    )

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id != 42:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

3. Custom Dependency with Error Handling

Dependencies in FastAPI can be used to encapsulate security logic, parameter validation, or any other preparatory steps that should happen before an endpoint function is called.

Copy

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

def verify_token(token: str):
    if token != "expected_token":
        raise HTTPException(status_code=403, detail="Invalid or expired token")
    return token

@app.get("/secure-data/")
async def secure_data(token: str = Depends(verify_token)):
    return {"message": "You have access to the secure data"}

4. Handling Errors in Background Tasks

Background tasks in FastAPI can also benefit from structured error handling, especially to avoid silent failures.

Copy

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    if not message:
        raise ValueError("No message provided")
    print(message)  # Simulate writing to a log file

def log_exception(task: BackgroundTasks):
    try:
        task.add_task(write_log, message="")
    except ValueError as e:
        return {"error": str(e)}

@app.post("/log/")
async def log_message(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, message="Log this message")
    return {"message": "Task is being logged"}

Real-World Considerations

• Extensive Use of HTTPException: For API endpoints, using HTTPException is common practice for immediately stopping request processing and responding with a specific HTTP status code.

• Logging: Ensure that all exceptions are logged properly. This helps in debugging and understanding the paths that lead to errors.

• User Feedback: Customize error messages to be helpful but avoid exposing sensitive details or implementation specifics that could be exploited.

These examples cover various scenarios of error handling in FastAPI, each tailored to different aspects of an application's needs, from simple validations to managing asynchronous tasks.

---

9

---

10

---

11

---

12

---

13

---

14

---

15

---

16

---

17

---

18

---

19

---

20