Our great sponsors
-
InfluxDB
Power Real-Time Data Analytics at Scale. Get real-time insights from all types of time series data with InfluxDB. Ingest, query, and analyze billions of data points in real-time with unbounded cardinality.
-
-
import logging import sys import time from datetime import datetime from typing import Any, Callable, TypeVar import uvicorn from bson import ObjectId from fastapi import FastAPI, HTTPException, Path, Request, Response from fastapi.middleware.cors import CORSMiddleware from motor.motor_asyncio import AsyncIOMotorClient from pydantic import BaseModel from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): mongo_uri: str root_path: str = "" logging_level: str = "INFO" model_config = SettingsConfigDict(env_file=".env", extra="ignore") settings = Settings() logging.basicConfig( stream=sys.stdout, level=settings.logging_level, format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s", # noqa: E501 datefmt="%d/%b/%Y %H:%M:%S", ) logger = logging.getLogger("my-todos") db_client = AsyncIOMotorClient(settings.mongo_uri) db = db_client.todoDb description = """ This is a fancy API built with [FastAPI🚀](https://fastapi.tiangolo.com/) 📝 [Source Code](https://github.com/dpills/fastapi-prod-guide) 🐞 [Issues](https://github.com/dpills/fastapi-prod-guide/issues) """ app = FastAPI( title="My Todo App", description=description, version="1.0.0", docs_url="/", root_path=settings.root_path, ) app.add_middleware( CORSMiddleware, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], allow_origins=[ "http://localhost:3000", ], ) MONGO_ID_REGEX = r"^[a-f\d]{24}$" F = TypeVar("F", bound=Callable[..., Any]) class Todo(BaseModel): title: str completed: bool = False class TodoId(BaseModel): id: str class TodoRecord(TodoId, Todo): created_date: datetime updated_date: datetime class NotFoundException(BaseModel): detail: str = "Not Found" @app.middleware("http") async def process_time_log_middleware(request: Request, call_next: F) -> Response: """ Add API process time in response headers and log calls """ start_time = time.time() response: Response = await call_next(request) process_time = str(round(time.time() - start_time, 3)) response.headers["X-Process-Time"] = process_time logger.info( "Method=%s Path=%s StatusCode=%s ProcessTime=%s", request.method, request.url.path, response.status_code, process_time, ) return response @app.post("/todos", response_model=TodoId) async def create_todo(payload: Todo) -> TodoId: """ Create a new Todo """ now = datetime.utcnow() insert_result = await db.todos.insert_one( { "title": payload.title, "completed": payload.completed, "created_date": now, "updated_date": now, } ) return TodoId(id=str(insert_result.inserted_id)) @app.get( "/todos/{id}", response_model=TodoRecord, responses={ 404: {"description": "Not Found", "model": NotFoundException}, }, ) async def get_todo( id: str = Path(description="Todo ID", pattern=MONGO_ID_REGEX) ) -> TodoRecord: """ Get a Todo """ doc = await db.todos.find_one({"_id": ObjectId(id)}) if not doc: raise HTTPException(status_code=404, detail="Not Found") return TodoRecord( id=str(doc["_id"]), title=doc["title"], completed=doc["completed"], created_date=doc["created_date"], updated_date=doc["updated_date"], ) @app.get("/todos", response_model=list[TodoRecord]) async def get_todos() -> list[TodoRecord]: """ Get Todos """ todos: list[TodoRecord] = [] async for doc in db.todos.find(): todos.append( TodoRecord( id=str(doc["_id"]), title=doc["title"], completed=doc["completed"], created_date=doc["created_date"], updated_date=doc["updated_date"], ) ) return todos @app.put( "/todos/{id}", response_model=TodoId, responses={ 404: {"description": "Not Found", "model": NotFoundException}, }, ) async def update_todo( payload: Todo, id: str = Path(description="Todo ID", pattern=MONGO_ID_REGEX), ) -> TodoId: """ Update a Todo """ now = datetime.utcnow() update_result = await db.todos.update_one( {"_id": ObjectId(id)}, { "$set": { "title": payload.title, "completed": payload.completed, "updated_date": now, } }, ) if update_result.matched_count == 0: raise HTTPException(status_code=404, detail="Not Found") return TodoId(id=id) @app.delete( "/todos/{id}", response_model=bool, responses={ 404: {"description": "Not Found", "model": NotFoundException}, }, ) async def delete_todo( id: str = Path(description="Todo ID", pattern=MONGO_ID_REGEX), ) -> bool: """ Delete a Todo """ delete_result = await db.todos.delete_one({"_id": ObjectId(id)}) if delete_result.deleted_count == 0: raise HTTPException(status_code=404, detail="Not Found") return True if __name__ == "__main__": uvicorn.run( "main:app", host="0.0.0.0", port=8000, log_level="debug", reload=True, )
Navigate to the GitHub Oauth Apps developer settings at https://github.com/settings/developers and create a new oauth app.
import hashlib from datetime import datetime import httpx from fastapi import APIRouter, HTTPException, Query from app.config import settings from app.utilities.db import db from .models import OauthException, OauthToken router = APIRouter() @router.get( "/callback", response_model=OauthToken, responses={ 400: {"description": "Oauth Error", "model": OauthException}, }, ) async def oauth_callback( code: str = Query(description="Authorization Code"), ) -> OauthToken: """ GitHub Oauth Integration Callback """ async with httpx.AsyncClient() as client: token_result = await client.post( "https://github.com/login/oauth/access_token", json={ "client_id": settings.github_oauth_client_id, "client_secret": settings.github_oauth_client_secret, "code": code, "redirect_uri": "http://localhost:8000/v1/auth/callback", }, headers={"Accept": "application/json"}, ) data = token_result.json() error = data.get("error") if error: raise HTTPException( status_code=400, detail=f"{data.get('error')}: {data.get('error_description')}", ) access_token: str = data.get("access_token") user_result = await client.get( "https://api.github.com/user", headers={"Authorization": f"Bearer {access_token}"}, ) user_data = user_result.json() user = user_data.get("login") await db.tokens.insert_one( { "user": user, "access_token_hash": hashlib.sha256(access_token.encode()).hexdigest(), "created_date": datetime.utcnow(), }, ) return OauthToken(access_token=access_token)
import logging import sys import time from datetime import datetime from typing import Any, Callable, TypeVar import uvicorn from bson import ObjectId from fastapi import FastAPI, HTTPException, Path, Request, Response from fastapi.middleware.cors import CORSMiddleware from motor.motor_asyncio import AsyncIOMotorClient from pydantic import BaseModel from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): mongo_uri: str root_path: str = "" logging_level: str = "INFO" model_config = SettingsConfigDict(env_file=".env", extra="ignore") settings = Settings() logging.basicConfig( stream=sys.stdout, level=settings.logging_level, format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s", # noqa: E501 datefmt="%d/%b/%Y %H:%M:%S", ) logger = logging.getLogger("my-todos") db_client = AsyncIOMotorClient(settings.mongo_uri) db = db_client.todoDb description = """ This is a fancy API built with [FastAPI🚀](https://fastapi.tiangolo.com/) 📝 [Source Code](https://github.com/dpills/fastapi-prod-guide) 🐞 [Issues](https://github.com/dpills/fastapi-prod-guide/issues) """ app = FastAPI( title="My Todo App", description=description, version="1.0.0", docs_url="/", root_path=settings.root_path, ) app.add_middleware( CORSMiddleware, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], allow_origins=[ "http://localhost:3000", ], ) MONGO_ID_REGEX = r"^[a-f\d]{24}$" F = TypeVar("F", bound=Callable[..., Any]) class Todo(BaseModel): title: str completed: bool = False class TodoId(BaseModel): id: str class TodoRecord(TodoId, Todo): created_date: datetime updated_date: datetime class NotFoundException(BaseModel): detail: str = "Not Found" @app.middleware("http") async def process_time_log_middleware(request: Request, call_next: F) -> Response: """ Add API process time in response headers and log calls """ start_time = time.time() response: Response = await call_next(request) process_time = str(round(time.time() - start_time, 3)) response.headers["X-Process-Time"] = process_time logger.info( "Method=%s Path=%s StatusCode=%s ProcessTime=%s", request.method, request.url.path, response.status_code, process_time, ) return response @app.post("/todos", response_model=TodoId) async def create_todo(payload: Todo) -> TodoId: """ Create a new Todo """ now = datetime.utcnow() insert_result = await db.todos.insert_one( { "title": payload.title, "completed": payload.completed, "created_date": now, "updated_date": now, } ) return TodoId(id=str(insert_result.inserted_id)) @app.get( "/todos/{id}", response_model=TodoRecord, responses={ 404: {"description": "Not Found", "model": NotFoundException}, }, ) async def get_todo( id: str = Path(description="Todo ID", pattern=MONGO_ID_REGEX) ) -> TodoRecord: """ Get a Todo """ doc = await db.todos.find_one({"_id": ObjectId(id)}) if not doc: raise HTTPException(status_code=404, detail="Not Found") return TodoRecord( id=str(doc["_id"]), title=doc["title"], completed=doc["completed"], created_date=doc["created_date"], updated_date=doc["updated_date"], ) @app.get("/todos", response_model=list[TodoRecord]) async def get_todos() -> list[TodoRecord]: """ Get Todos """ todos: list[TodoRecord] = [] async for doc in db.todos.find(): todos.append( TodoRecord( id=str(doc["_id"]), title=doc["title"], completed=doc["completed"], created_date=doc["created_date"], updated_date=doc["updated_date"], ) ) return todos @app.put( "/todos/{id}", response_model=TodoId, responses={ 404: {"description": "Not Found", "model": NotFoundException}, }, ) async def update_todo( payload: Todo, id: str = Path(description="Todo ID", pattern=MONGO_ID_REGEX), ) -> TodoId: """ Update a Todo """ now = datetime.utcnow() update_result = await db.todos.update_one( {"_id": ObjectId(id)}, { "$set": { "title": payload.title, "completed": payload.completed, "updated_date": now, } }, ) if update_result.matched_count == 0: raise HTTPException(status_code=404, detail="Not Found") return TodoId(id=id) @app.delete( "/todos/{id}", response_model=bool, responses={ 404: {"description": "Not Found", "model": NotFoundException}, }, ) async def delete_todo( id: str = Path(description="Todo ID", pattern=MONGO_ID_REGEX), ) -> bool: """ Delete a Todo """ delete_result = await db.todos.delete_one({"_id": ObjectId(id)}) if delete_result.deleted_count == 0: raise HTTPException(status_code=404, detail="Not Found") return True if __name__ == "__main__": uvicorn.run( "main:app", host="0.0.0.0", port=8000, log_level="debug", reload=True, )
Whenever I start a new project I like to maintain quality standards and using automated quality tools makes it easy. Lets go ahead and install mypy for static type checking, black for formatting, and ruff for linting. Add these to the dev dependencies.
Whenever I start a new project I like to maintain quality standards and using automated quality tools makes it easy. Lets go ahead and install mypy for static type checking, black for formatting, and ruff for linting. Add these to the dev dependencies.