FastAPI with MooseStack

from fastapi import FastAPI, Request, HTTPExceptionfrom moose_lib.dmv2 import WebApp, WebAppConfig, WebAppMetadatafrom moose_lib.dmv2.web_app_helpers import get_moose_utilsfrom app.tables.my_table import MyTable app = FastAPI() @app.get("/health")async def health():    return {"status": "ok"} @app.get("/data")async def get_data(request: Request, limit: int = 10):    moose = get_moose_utils(request)    if not moose:        raise HTTPException(            status_code=500,            detail="Moose utilities not available"        )     try:        query = f"""            SELECT                 {MyTable.columns.id},                {MyTable.columns.name},                {MyTable.columns.created_at}            FROM {MyTable}            ORDER BY {MyTable.columns.created_at} DESC            LIMIT {{limit}}        """         result = moose.client.query.execute_raw(query, {            "limit": limit        })         return {"success": True, "data": result}    except Exception as error:        raise HTTPException(status_code=500, detail=str(error)) # Register as WebAppfastapi_app = WebApp(    "fastApiApp",    app,    WebAppConfig(        mount_path="/fastapi",        metadata=WebAppMetadata(description="FastAPI application")    ))

from fastapi import FastAPI, Request, HTTPException, Depends, BackgroundTasksfrom fastapi.responses import JSONResponsefrom fastapi.middleware.cors import CORSMiddlewarefrom moose_lib.dmv2 import WebApp, WebAppConfig, WebAppMetadatafrom moose_lib.dmv2.web_app_helpers import get_moose_utils, ApiUtilfrom app.tables.user_events import UserEventsfrom app.tables.user_profile import UserProfilefrom pydantic import BaseModel, Fieldfrom datetime import datetimefrom typing import Optional app = FastAPI(    title="Advanced API",    description="Advanced FastAPI application with MooseStack",    version="1.0.0") # CORS middlewareapp.add_middleware(    CORSMiddleware,    allow_origins=["*"],    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],) # Custom middleware@app.middleware("http")async def log_requests(request: Request, call_next):    start_time = datetime.now()    response = await call_next(request)    duration = (datetime.now() - start_time).total_seconds()    print(f"{request.method} {request.url.path} - {duration:.3f}s")    return response # Request/Response modelsclass EventQuery(BaseModel):    limit: int = Field(10, ge=1, le=100)    event_type: Optional[str] = None class EventData(BaseModel):    event_type: str = Field(..., min_length=1)    data: dict class EventResponse(BaseModel):    id: str    event_type: str    timestamp: datetime # Health check@app.get("/health")async def health():    return {        "status": "ok",        "timestamp": datetime.now().isoformat()    } # GET with path and query parameters@app.get("/users/{user_id}/events", response_model=dict)async def get_user_events(    request: Request,    user_id: str,    limit: int = 10,    event_type: Optional[str] = None):    moose = get_moose_utils(request)    if not moose:        raise HTTPException(status_code=500, detail="Moose utilities not available")     query = """        SELECT            id,            event_type,            timestamp        FROM {table}        WHERE user_id = {user_id}        {event_filter}        ORDER BY timestamp DESC        LIMIT {limit}    """     event_filter = "AND event_type = {event_type}" if event_type else ""    params = {        "table": UserEvents,        "user_id": user_id,        "limit": limit    }    if event_type:        params["event_type"] = event_type     try:        result = moose.client.query.execute(            query.format(event_filter=event_filter),            params        )        return {            "user_id": user_id,            "count": len(result),            "events": result        }    except Exception as error:        raise HTTPException(status_code=500, detail=str(error)) # POST with validated body@app.post("/users/{user_id}/events", status_code=201)async def create_event(    request: Request,    user_id: str,    body: EventData,    background_tasks: BackgroundTasks):    moose = get_moose_utils(request)    if not moose:        raise HTTPException(status_code=500, detail="Moose utilities not available")     # Background task    def log_event_creation(user_id: str, event_type: str):        print(f"Event created: {user_id} - {event_type}")     background_tasks.add_task(log_event_creation, user_id, body.event_type)     return {        "success": True,        "user_id": user_id,        "event_type": body.event_type,        "data": body.data    } # Protected endpoint with dependency injectionasync def require_auth(request: Request) -> ApiUtil:    moose = get_moose_utils(request)    if not moose or not moose.jwt:        raise HTTPException(status_code=401, detail="Unauthorized")    return moose @app.get("/protected")async def protected(moose: ApiUtil = Depends(require_auth)):    return {        "message": "Authenticated",        "user": moose.jwt.get("sub"),        "claims": moose.jwt    } # Admin endpoint with role checkasync def require_admin(moose: ApiUtil = Depends(require_auth)) -> ApiUtil:    role = moose.jwt.get("role")    if role != "admin":        raise HTTPException(status_code=403, detail="Forbidden")    return moose @app.get("/admin/stats")async def admin_stats(moose: ApiUtil = Depends(require_admin)):    return {"message": "Admin access granted"} # WebSocket supportfrom fastapi import WebSocket @app.websocket("/ws")async def websocket_endpoint(websocket: WebSocket):    await websocket.accept()    try:        while True:            data = await websocket.receive_text()            await websocket.send_text(f"Echo: {data}")    except Exception:        await websocket.close() # Global exception handler@app.exception_handler(Exception)async def global_exception_handler(request: Request, exc: Exception):    return JSONResponse(        status_code=500,        content={"error": "Internal Server Error", "message": str(exc)}    ) # Register as WebAppfastapi_app = WebApp(    "advancedFastApi",    app,    WebAppConfig(        mount_path="/api/v1",        metadata=WebAppMetadata(            description="Advanced FastAPI application with middleware and dependencies"        )    ))

WebApp(name, app, config)

@dataclassclass WebAppConfig:    mount_path: str                              # Required: URL path    metadata: Optional[WebAppMetadata] = None    # Optional: Documentation    inject_moose_utils: bool = True              # Optional: Inject utilities @dataclassclass WebAppMetadata:    description: Optional[str] = None

from moose_lib.dmv2.web_app_helpers import get_moose_utils @app.get("/data")async def get_data(request: Request):    moose = get_moose_utils(request)    if not moose:        raise HTTPException(status_code=500, detail="Utilities not available")     client = moose.client    jwt = moose.jwt

from fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddlewarefrom fastapi.middleware.gzip import GZipMiddlewarefrom starlette.middleware.sessions import SessionMiddleware app = FastAPI() # CORSapp.add_middleware(    CORSMiddleware,    allow_origins=["https://example.com"],    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],) # Compressionapp.add_middleware(GZipMiddleware, minimum_size=1000) # Sessionsapp.add_middleware(SessionMiddleware, secret_key="your-secret-key") # Custom middleware@app.middleware("http")async def add_custom_header(request: Request, call_next):    response = await call_next(request)    response.headers["X-Custom-Header"] = "Value"    return response

from fastapi import Depends, HTTPExceptionfrom typing import Optional # Auth dependencyasync def get_current_user(moose: ApiUtil = Depends(get_moose_dependency())) -> dict:    if not moose.jwt:        raise HTTPException(status_code=401, detail="Not authenticated")     user_id = moose.jwt.get("sub")    # Fetch user from database    return {"id": user_id} # Admin dependencyasync def require_admin_user(user: dict = Depends(get_current_user)) -> dict:    if user.get("role") != "admin":        raise HTTPException(status_code=403, detail="Not authorized")    return user # Use dependencies@app.get("/user/profile")async def get_profile(user: dict = Depends(get_current_user)):    return user @app.get("/admin/dashboard")async def admin_dashboard(user: dict = Depends(require_admin_user)):    return {"message": f"Welcome admin {user['id']}"}

from fastapi import Dependsfrom app.tables.user_table import UserTable class Pagination:    def __init__(self, page: int = 1, size: int = 10):        self.page = page        self.size = size        self.skip = (page - 1) * size @app.get("/users")async def list_users(    pagination: Pagination = Depends(),    moose: ApiUtil = Depends(get_moose_dependency())):    # Using parameterized query for user input values    query = f"""        SELECT             {UserTable.columns.id},            {UserTable.columns.name},            {UserTable.columns.email}        FROM {UserTable}        WHERE {UserTable.columns.status} = 'active'        LIMIT {{size:UInt32}}        OFFSET {{skip:UInt32}}    """    result = moose.client.query.execute_raw(query, {        "size": pagination.size,        "skip": pagination.skip    })    return result

from pydantic import BaseModel, Field, validatorfrom datetime import datetimefrom typing import Optional, Literal class UserEventCreate(BaseModel):    event_type: Literal["click", "view", "purchase"]    timestamp: datetime = Field(default_factory=datetime.now)    properties: dict = Field(default_factory=dict)    value: Optional[float] = Field(None, ge=0, le=1000000)     @validator('properties')    def validate_properties(cls, v):        if len(v) > 50:            raise ValueError('Too many properties')        return v @app.post("/events")async def create_event(    event: UserEventCreate,    moose: ApiUtil = Depends(get_moose_dependency())):    # event is fully validated    return {"success": True, "event": event}

from fastapi import BackgroundTasks def send_notification(user_id: str, message: str):    # Expensive operation    print(f"Sending notification to {user_id}: {message}") @app.post("/notify/{user_id}")async def notify_user(    user_id: str,    message: str,    background_tasks: BackgroundTasks):    # Add task to run after response is sent    background_tasks.add_task(send_notification, user_id, message)    return {"message": "Notification queued"}

# Manual check@app.get("/protected")async def protected(request: Request):    moose = get_moose_utils(request)    if not moose or not moose.jwt:        raise HTTPException(status_code=401, detail="Unauthorized")     user_id = moose.jwt.get("sub")    return {"message": "Authenticated", "user_id": user_id} # Dependency pattern (recommended)async def require_auth(request: Request) -> ApiUtil:    moose = get_moose_utils(request)    if not moose or not moose.jwt:        raise HTTPException(status_code=401, detail="Unauthorized")    return moose @app.get("/protected")async def protected(moose: ApiUtil = Depends(require_auth)):    return {"message": "Authenticated", "user": moose.jwt.get("sub")}

from fastapi import APIRouter, Dependsfrom moose_lib.dmv2.web_app_helpers import get_moose_dependency, ApiUtil router = APIRouter(prefix="/users", tags=["users"]) @router.get("/{user_id}")async def get_user(    user_id: str,    moose: ApiUtil = Depends(get_moose_dependency())):    return {"user_id": user_id}

from fastapi import FastAPIfrom moose_lib.dmv2 import WebApp, WebAppConfigfrom .routers import users app = FastAPI() # Include routersapp.include_router(users.router) webapp = WebApp("mainApp", app, WebAppConfig(mount_path="/api"))

WebAppConfig(mount_path="/myapi", inject_moose_utils=True)

# Correctapp = FastAPI()webapp = WebApp("name", app, config) # Incorrectrouter = APIRouter()webapp = WebApp("name", router, config)  # Won't work

# Correctasync def get_moose(request: Request) -> ApiUtil:    moose = get_moose_utils(request)    if not moose:        raise HTTPException(500, "Not available")    return moose  # Return ApiUtil # Usage@app.get("/")async def handler(moose: ApiUtil = Depends(get_moose)):    # moose is ApiUtil type    pass