FastAPI

pip install fastapi uvicorn

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

uvicorn main:app --reload

uvicorn main:app --workers 4

@app.get("/items/")
  def read_items():
      return [{"name": "Item 1"}, {"name": "Item 2"}]

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

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

@app.get("/files/{file_path:path}")
  def read_file(file_path: str):
      return {"file_path": file_path}

@app.get("/items/")
  def read_items(skip: int = 0, limit: int = 10):
      return db.items[skip: skip + limit]

@app.get("/items/{item_id}")
  def read_item(item_id: str, q: Optional[str] = None):
      return {"item_id": item_id, "q": q}

from pydantic import BaseModel
  
  class Item(BaseModel):
      name: str
      description: Optional[str] = None
      price: float
      tax: Optional[float] = None

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

@app.post("/items/", response_model=Item)
  def create_item(item: Item):
      return item

from fastapi import status
  
  @app.post("/items/", status_code=status.HTTP_201_CREATED)
  def create_item(item: Item):
      return item

async def common_parameters(q: Optional[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

async def get_db():
      db = SessionLocal()
      try:
          yield db
      finally:
          db.close()
  
  @app.post("/users/")
  def create_user(user: UserCreate, db: Session = Depends(get_db)):
      db_user = User(**user.dict())
      db.add(db_user)
      db.commit()
      db.refresh(db_user)
      return db_user

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

from jose import JWTError, jwt
  
  SECRET_KEY = "secret"
  ALGORITHM = "HS256"
  
  def create_access_token(data: dict):
      to_encode = data.copy()
      return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

from fastapi import FastAPI, HTTPException
  
  @app.exception_handler(HTTPException)
  async def http_exception_handler(request, exc):
      return JSONResponse(
          status_code=exc.status_code,
          content={"detail": exc.detail},
      )

from fastapi import HTTPException
  
  @app.get("/items/{item_id}")
  def read_item(item_id: str):
      item = db.get(item_id)
      if not item:
          raise HTTPException(status_code=404, detail="Item not found")
      return item

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

@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
      logger.info(f"{request.method} {request.url} {response.status_code} {process_time:.2f}s")
      return response

from fastapi import BackgroundTasks
  
  def write_notification(email: str, message=""):
      # Send email in background
      send_email(email, message)
  
  @app.post("/send-notification/{email}")
  async def send_notification(email: str, background_tasks: BackgroundTasks):
      background_tasks.add_task(write_notification, email, message="Hello!")
      return {"message": "Notification sent in background"}

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

from sqlalchemy import create_engine
  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()

# Initialize Alembic
  alembic init migrations
  
  # Create new migration
  alembic revision --autogenerate -m "Add users table"
  
  # Apply migrations
  alembic upgrade head

from fastapi import File, UploadFile
  
  @app.post("/upload/")
  async def upload_file(file: UploadFile = File(...)):
      contents = await file.read()
      save_file(file.filename, contents)
      return {"filename": file.filename}

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

from fastapi import WebSocket
  
  @app.websocket("/ws")
  async def websocket_endpoint(websocket: WebSocket):
      await websocket.accept()
      while True:
          data = await websocket.receive_text()
          await websocket.send_text(f"Message: {data}")

app = FastAPI(
      title="My API",
      description="API for my application",
      version="0.1.0",
      openapi_url="/api/v1/openapi.json",
      docs_url="/docs",
      redoc_url="/redoc",
  )

FROM python:3.9
  
  WORKDIR /app
  COPY requirements.txt .
  RUN pip install -r requirements.txt
  COPY . .
  
  CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

uvicorn main:app --ssl-keyfile key.pem --ssl-certfile cert.pem

from fastapi_cache import FastAPICache
  from fastapi_cache.backends.redis import RedisBackend
  from fastapi_cache.decorator import cache
  
  @app.on_event("startup")
  async def startup():
      redis = aioredis.from_url("redis://localhost")
      FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
  
  @app.get("/")
  @cache(expire=60)
  async def index():
      return dict(hello="world")

from fastapi.routing import APIRoute
  
  class CustomRoute(APIRoute):
      def get_route_handler(self):
          original_route_handler = super().get_route_handler()
          
          async def custom_route_handler(request):
              # Pre-processing
              response = await original_route_handler(request)
              # Post-processing
              return response
              
          return custom_route_handler
  
  app = FastAPI()
  app.router.route_class = CustomRoute