codesnips

fastapi

python 
import datetime as dt

from sqlalchemy import Column, DateTime, Integer, String, UniqueConstraint
from sqlalchemy.orm import declarative_base

Base = declarative_base()

Idempotent Webhook Ingestion With a Postgres Dedupe Store in FastAPI

fastapi webhooks idempotency
by codesnips 3 tabs
python 
from datetime import datetime, timedelta, timezone

from fastapi.security import OAuth2PasswordBearer
from jose import jwt
from passlib.context import CryptContext

Reusable OAuth2 Password Bearer JWT Authentication Dependency in FastAPI

fastapi oauth2 jwt
by codesnips 4 tabs
python 
from datetime import datetime, timedelta, timezone

from jose import jwt, JWTError
from jose.exceptions import ExpiredSignatureError

SECRET_KEY = "change-me-in-production"

FastAPI JWT Authentication with Access/Refresh Tokens and a Verification Dependency

fastapi jwt authentication
by codesnips 3 tabs
python 
import base64
import json
from datetime import datetime
from typing import NamedTuple

from fastapi import HTTPException

Cursor-Based Pagination in FastAPI with a Dependency and RFC 5988 Link Headers

fastapi pagination cursor
by codesnips 3 tabs
python 
import asyncio
import uuid
from dataclasses import dataclass, field
from typing import Any, Dict, Optional

Long-Polling a Background Job's Status with a Shared Result Registry in FastAPI

fastapi asyncio long-polling
by codesnips 3 tabs
lua 
-- KEYS[1] = bucket key
-- ARGV[1] = capacity, ARGV[2] = refill_per_sec, ARGV[3] = now (float seconds)
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

Redis-Backed Token Bucket Rate Limiter as FastAPI Middleware

fastapi redis rate-limiting
by codesnips 4 tabs
python 
import enum
import datetime as dt

from sqlalchemy import Column, Integer, String, BigInteger, Enum, DateTime
from sqlalchemy.orm import declarative_base

Streaming FastAPI File Uploads with a Background Validation Task

fastapi uploads background-tasks
by codesnips 3 tabs
python 
import hmac
import hashlib
import time
from fastapi import Request, HTTPException, status

Verifying Stripe Webhook Signatures With a Reusable FastAPI Dependency

fastapi webhooks security
by codesnips 3 tabs
python 
import asyncio
from typing import Awaitable, Callable, Dict, TypeVar

T = TypeVar("T")

Coalesce Duplicate In-Flight Requests with an Asyncio Single-Flight Cache

asyncio concurrency caching
by codesnips 3 tabs
python 
from enum import Enum
from typing import Optional

from fastapi import Query
from pydantic import BaseModel, Field, field_validator

Validate and Normalize Query Filters With a Pydantic Dependency in FastAPI

fastapi pydantic validation
by codesnips 3 tabs