FastAPI

# пример приложения FastAPI
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None

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

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/")
async def read_root():
    await asyncio.sleep(1)
    return {"Hello": "World"}

# Пример базовой структуры FastAPI-приложения
from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

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

def common_parameters(q: str = None):
    return q

@app.get("/items/")
async def read_items(q: str = Depends(common_parameters)):
    return {"q": q}

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

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.post("/items/")
async def create_item(item: Item):
    return {"name": item.name, "price": item.price}

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False

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

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str

data = {"id": "123", "name": "Alice"}
user = User(**data)
print(user)

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="API documentation with automatic docs generation via Swagger UI and ReDoc",
    version="1.0.0",
)

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

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi import FastAPI

app = FastAPI()

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

from fastapi import FastAPI

app = FastAPI()

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

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str

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

from fastapi import FastAPI
app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello World"}

@app.post("/items")
def create_item(item: dict):
    return {"item": item}

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

@app.patch("/items/{item_id}")
def partial_update_item(item_id: int, item: dict):
    return {"item_id": item_id, "item": item}

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    return {"item_id": item_id, "deleted": True}

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class MyCustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(MyCustomException)
async def my_custom_exception_handler(request: Request, exc: MyCustomException):
    return JSONResponse(
        status_code=418,
        content={"message": f"An error occurred: {exc.name}"},
    )

@app.get("/")
async def read_main():
    raise MyCustomException("Something went wrong")

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"message": "Internal Server Error"}
    )

@app.get("/")
async def read_root():
    raise Exception("Test error")

from fastapi import FastAPI

app = FastAPI()

# Синхронный маршрут
@app.get("/sync")
def sync_route():
    return {"message": "Hello from sync endpoint!"}

# Асинхронный маршрут
@app.get("/async")
async def async_route():
    return {"message": "Hello from async endpoint!"}

import asyncio
import time

def sync_blocking():
    time.sleep(2)
    return 'blocking result'

async def async_handler():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, sync_blocking)
    return result

if __name__ == '__main__':
    result = asyncio.run(async_handler())
    print(result)

# Пример реализации Dependency Injection в FastAPI
from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = "database connection"
    try:
        yield db
    finally:
        print("Закрытие соединения с базой данных")

@app.get("/items/")
def read_items(db: str = Depends(get_db)):
    return {"db": db}

# Пример системы зависимостей с использованием инверсии управления
class Database:
    def query(self):
        return "Данные из базы"

class Service:
    def __init__(self, db):
        self.db = db
    def get_data(self):
        return self.db.query()

# Тестовый подмена зависимости
class FakeDatabase:
    def query(self):
        return "Фейковые данные"

# Использование зависимости для тестирования
db = FakeDatabase()
service = Service(db)
print(service.get_data())

from fastapi import FastAPI, Header, HTTPException, status, Depends

app = FastAPI()

def verify_token(x_token: str = Header(...)):
    if x_token != "SuperSecretToken":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing token"
        )
    return x_token

@app.get("/protected")
def protected_route(token: str = Depends(verify_token)):
    return {"message": "Protected content"}

from fastapi import Depends, FastAPI, HTTPException

def verify_user_data(user_data: str):
    if user_data != "expected_value":
         raise HTTPException(status_code=400, detail="Неверное значение")
    return user_data

app = FastAPI()

@app.get("/items/")
async def read_item(user: str = Depends(verify_user_data)):
    return {"user": user}

# Пример использования аннотаций типов в FastAPI
from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

app = FastAPI()

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

from flask import Flask, jsonify
app = Flask(__name__)
@app.route("/user/<int:user_id>/post/<int:post_id>", methods=["GET"])
def get_post(user_id, post_id):
    return jsonify({"user_id": user_id, "post_id": post_id})
if __name__ == "__main__":
    app.run(debug=True)

from fastapi import FastAPI, Form, File, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload_form(
    username: str = Form("default_user"), 
    file: UploadFile = File(...)
):
    return {"username": username, "filename": file.filename}

from flask import Flask, request, jsonify

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # Ограничение на 16 МБ

@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return jsonify({'error': 'No file part'}), 400
    file = request.files['file']
    if file.filename == '':
        return jsonify({'error': 'No selected file'}), 400
    # Дополнительная валидация типа файла может быть здесь
    file.save(f'/path/to/save/{file.filename}')
    return jsonify({'message': 'File uploaded successfully'})

if __name__ == '__main__':
    app.run(debug=True)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

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

app = FastAPI()

@app.post("/items/")
async def create_item(item: Item):
    if item.price < 0:
        raise HTTPException(status_code=400, detail="Price must be positive")
    return item

from pydantic import BaseModel

class UserResponse(BaseModel):
    id: int
    name: str

@app.get("/user/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    user = get_user_from_db(user_id)
    return user

from fastapi import FastAPI
from pydantic import BaseModel

class User(BaseModel):
    name: str = "Anonymous"
    age: int = 0

app = FastAPI()

@app.get("/user", response_model=User, response_model_exclude_unset=True)
def get_user():
    return User(name="Alice")

def serialize_data(data, allowed_fields):
    # Фильтрация данных с использованием белого списка
    return {field: data.get(field) for field in allowed_fields}

user_data = {
    "id": 123,
    "username": "john",
    "password": "secret",  # чувствительные данные
    "email": "john@example.com",
}

allowed_fields = ["id", "username", "email"]
serialized = serialize_data(user_data, allowed_fields)
print(serialized)

import datetime
from fastapi import FastAPI
from pydantic import BaseModel

def custom_datetime_serializer(value: datetime.datetime) -> str:
    return value.strftime("%Y-%m-%d %H:%M:%S")

class Item(BaseModel):
    id: int
    name: str
    created_at: datetime.datetime

    class Config:
        json_encoders = {
            datetime.datetime: custom_datetime_serializer
        }

app = FastAPI()

@app.get("/item")
def get_item():
    item = Item(id=1, name="Test Item", created_at=datetime.datetime.now())
    return item.dict()

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class CustomMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Логика до вызова основного обработчика
        response = await call_next(request)
        # Логика после вызова основного обработчика
        return response

app.add_middleware(CustomMiddleware)

from fastapi import FastAPI, Request
import logging
from time import time

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time()
    response = await call_next(request)
    process_time = time() - start_time
    logging.info(f"{request.method} {request.url} completed in {process_time:.2f}s")
    return response

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

from flask import Flask, jsonify
from flask_cors import CORS

app = Flask("__name__")
# Указываем конкретный origin вместо "*"
CORS(app, origins=["http://example.com"], supports_credentials=True)

@app.route("/data")
def data():
    return jsonify({"message": "Hello, CORS!"})

if __name__ == "__main__":
    app.run()

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str) -> None:
    with open("log.txt", "a") as log_file:
        log_file.write(message + "\n")

@app.post("/send-data")
async def send_data(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "Data sent")
    return {"status": "processing"}

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

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

@app.get("/send-notification")
def send_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "Notification sent")
    return {"message": "Notification will be sent in background"}

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379/0")

@app.task
def long_process(x, y):
    return x + y

import uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@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"Message received: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

import asyncio
import websockets

async def handler(websocket, path):
    try:
        async for message in websocket:
            # Обработка входящих сообщений
            await websocket.send(f"Echo: {message}")
    except websockets.exceptions.ConnectionClosed as e:
        # Логирование ошибки и закрытие соединения
        print(f"Соединение закрыто: {e}")

start_server = websockets.serve(handler, "localhost", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, list[WebSocket]] = {}

    async def connect(self, group: str, websocket: WebSocket):
        await websocket.accept()
        if group not in self.active_connections:
            self.active_connections[group] = []
        self.active_connections[group].append(websocket)

    def disconnect(self, group: str, websocket: WebSocket):
        self.active_connections[group].remove(websocket)
        if len(self.active_connections[group]) == 0:
            del self.active_connections[group]

    async def send_message_to_group(self, group: str, message: str):
        if group in self.active_connections:
            for connection in self.active_connections[group]:
                await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{group}")
async def websocket_endpoint(websocket: WebSocket, group: str):
    await manager.connect(group, websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_message_to_group(group, data)
    except WebSocketDisconnect:
        manager.disconnect(group, websocket)

from fastapi import FastAPI, WebSocket

app = FastAPI()

@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 text was: {data}")

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient

app = FastAPI()

def get_text():
    return "Hello, World!"

@app.get("/greet")
def greet(text: str = Depends(get_text)):
    return {"message": text}

def override_get_text():
    return "Test Hello!"

app.dependency_overrides[get_text] = override_get_text

client = TestClient(app)

def test_greet():
    response = client.get("/greet")
    assert response.status_code == 200
    assert response.json() == {"message": "Test Hello!"}

import pytest
from fastapi import FastAPI
from httpx import AsyncClient

app = FastAPI()

@app.get("/async-endpoint")
async def async_endpoint():
    return {"message": "success"}

@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/async-endpoint")
        assert response.status_code == 200
        assert response.json() == {"message": "success"}

# Пример использования Dependency Injection с моком зависимости

class Service:
    def __init__(self, dependency):
        self.dependency = dependency

    def perform_action(self):
        return self.dependency.execute()

class Dependency:
    def execute(self):
        return "Real dependency"

class MockDependency:
    def execute(self):
        return "Mocked dependency"

# Внедрение зависимости через конструктор
def main():
    # Замена зависимости на мок для тестирования
    service = Service(MockDependency())
    result = service.perform_action()
    print(result)

if __name__ == "__main__":
    main()

import logging

# Настройка базового логирования
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

logger = logging.getLogger(__name__)

logger.info("Логирование запущено")

from loguru import logger

# Настройка логирования с Loguru
logger.add("logfile.log", rotation="500 MB")
logger.info("Приложение запущено")

import logging

logging.basicConfig(level=logging.DEBUG,
                    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    logger.debug("Debug message: Root endpoint accessed")
    # Для дебаггинга можно использовать pdb
    # import pdb; pdb.set_trace()
    return {"Hello": "World"}

import uvicorn
from fastapi import FastAPI
from starlette_exporter import PrometheusMiddleware, handle_metrics

app = FastAPI()
app.add_middleware(PrometheusMiddleware)
app.add_route("/metrics", handle_metrics)

if __name__ == "__main__":
    uvicorn.run(app)

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
from models import Parent

async def get_parents(session: AsyncSession):
    result = await session.execute(
        select(Parent).options(selectinload(Parent.children))
    )
    return result.scalars().all()

# Импорт необходимых модулей с асинхронным расширением SQLAlchemy
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# Создание асинхронного движка
engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname", echo=True)

# Создание фабрики сессий
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_data():
    # Правильное использование контекстного менеджера для сессии
    async with async_session() as session:
        async with session.begin():
            result = await session.execute("SELECT * FROM table")
            data = result.fetchall()
            return data

# Неправильное: использование синхронной сессии в асинхронной функции может блокировать event loop
def get_data_sync():
    session = async_session()  # Ошибка: использована синхронная логика
    result = session.execute("SELECT * FROM table")
    data = result.fetchall()
    return data

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

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

app = FastAPI()

@app.get("/items/")
async def read_items():
    async with async_session() as session:
        result = await session.execute("SELECT * FROM items")
        items = result.fetchall()
        return items

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

app = FastAPI()
SQLALCHEMY_DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(
    SQLALCHEMY_DATABASE_URL,
    pool_size=10,
    max_overflow=20
)

async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

@app.get("/")
async def read_root():
    async with async_session() as session:
        result = await session.execute("SELECT 1")
        return {"result": result.scalar()}

# Sample code
from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

# Presentation layer
@app.get("/items/{item_id}")
async def read_item(item_id: int, service: ItemService = Depends(get_item_service)):
    return await service.get_item(item_id)

# Business logic layer
class ItemService:
    async def get_item(self, item_id: int):
        # Business logic here
        return {"item_id": item_id, "data": "Details about the item"}

def get_item_service():
    return ItemService()

# file: app/main.py
from fastapi import FastAPI
from app.api import router as api_router
from app.core.config import settings

app = FastAPI(title=settings.PROJECT_NAME)

app.include_router(api_router, prefix="/api")

# file: app/api.py
from fastapi import APIRouter
from app.endpoints import users, items

router = APIRouter()
router.include_router(users.router, prefix="/users", tags=['Users'])
router.include_router(items.router, prefix="/items", tags=['Items'])

# file: app/endpoints/users.py
from fastapi import APIRouter
from app.schemas.user import UserCreate, UserOut
from app.services.user_service import create_user

router = APIRouter()

@router.post("/", response_model=UserOut)
def register_user(user: UserCreate):
    return create_user(user)

from fastapi import Depends, FastAPI

class UserRepository:
    def get_user(self, user_id: int):
        # Реализуйте логику получения пользователя из базы данных
        return {"user_id": user_id, "name": "John Doe"}

class UnitOfWork:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    def commit(self):
        # Здесь логика сохранения изменений
        pass

app = FastAPI()

def get_repository():
    return UserRepository()

@app.get("/users/{user_id}")
def read_user(user_id: int, repo: UserRepository = Depends(get_repository)):
    user = repo.get_user(user_id)
    return user

from fastapi import FastAPI

app = FastAPI()

@app.get("/v1/items")
def read_items_v1():
    return {"message": "Версия 1"}

@app.get("/v2/items")
def read_items_v2():
    return {"message": "Версия 2"}

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def authenticate_user(username: str, password: str):
    # В реальном проекте проверка осуществляется через базу данных
    if username == "johndoe" and password == "secret":
        return {"username": username}
    return None

def create_access_token(data: dict):
    # Пример генерации JWT токена с алгоритмом HS256
    return jwt.encode(data, "secret", algorithm="HS256")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Неверные учетные данные",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user["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", algorithms=["HS256"])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Неверный токен")
    except jwt.PyJWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Неверный токен")
    return {"username": username}

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

SECRET_KEY = "secret"
ALGORITHM = "HS256"

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # Здесь должна быть проверка пользователя (например, из базы данных)
    token_data = {"sub": form_data.username}
    token = jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM)
    return {"access_token": 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])
        user = payload.get("sub")
        if user is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
    except jwt.PyJWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
    return {"user": user}

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class User(BaseModel):
    username: str
    is_active: bool = True

def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    # Кастомная логика проверки токена
    if token != "secrettoken":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    user = User(username="testuser")
    return user

@app.get("/protected-endpoint")
def protected_endpoint(current_user: User = Depends(get_current_user)):
    return {"user": current_user.username}

from flask import Flask, request, jsonify
from flask_wtf.csrf import CSRFProtect, validate_csrf

app = Flask("app")
app.config["SECRET_KEY"] = "your-secret-key"
csrf = CSRFProtect(app)

@app.route("/api/data", methods=["POST"])
def api_data():
    try:
        token = request.headers.get("X-CSRFToken")
        validate_csrf(token)
    except Exception as e:
        return jsonify({"error": "CSRF validation failed"}), 400
    # обработка данных запроса
    return jsonify({"message": "Data processed successfully"})

if __name__ == "__main":
    app.run()

from pydantic import BaseSettings

class Settings(BaseSettings):
    secret_key: str
    database_url: str

    class Config:
        env_file = ".env"

settings = Settings()

from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI Application"
    debug: bool = True

    class Config:
        env_file = ".env"

settings = Settings()
print(settings.app_name)

from pydantic import BaseSettings

class Settings(BaseSettings):
    app_name: str = "My FastAPI Application"
    items_per_user: int = 50

    class Config:
        env_file = ".env"

settings = Settings()

from fastapi import FastAPI

app = FastAPI()

@app.get("/config")
def get_config():
    return {
        "app_name": settings.app_name,
        "items_per_user": settings.items_per_user
    }

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app:app", host="0.0.0.0", port=8000, workers=4)

# Пример запуска FastAPI с Uvicorn:
import uvicorn

if __name__ == "__main__":
    uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=False)

# Для Gunicorn команда запуска:
# gunicorn -k uvicorn.workers.UvicornWorker app:app

# Для Hypercorn команда запуска:
# hypercorn app:app --bind 0.0.0.0:8000

# <Пример конфигурации Nginx для FastAPI>
server {
    listen 80;
    server_name example.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_redirect off;
    }
}

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, world!"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, workers=4)

from functools import lru_cache

@lru_cache(maxsize=128)
def compute_data(n):
    # Имитация затратной операции
    result = n * n
    return result

if __name__ == '__main__':
    print(compute_data(10))

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from fastapi.responses import JSONResponse

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)

@app.get("/limited")
@limiter.limit("5/minute")
async def limited_endpoint(request: Request):
    return {"message": "OK"}

import asyncio
from fastapi import FastAPI

app = FastAPI()

def blocking_task():
    import time
    time.sleep(5)
    return "done"

@app.get("/")
async def main():
    result = await asyncio.to_thread(blocking_task)
    return {"result": result}

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def io_bound_task():
    # I/O-bound: асинхронное выполнение
    await asyncio.sleep(1)
    return "I/O результат"

def cpu_intensive_task(data):
    # CPU-интенсивное вычисление
    result = 0
    for i in range(10**7):
        result += i
    return result

async def main():
    # Выполнение I/O-bound задачи
    io_result = await io_bound_task()
    # Выполнение CPU-интенсивной задачи в отдельном процессе
    loop = asyncio.get_running_loop()
    cpu_result = await loop.run_in_executor(ProcessPoolExecutor(), cpu_intensive_task, io_result)
    print("I/O:", io_result, "CPU:", cpu_result)

asyncio.run(main())

from fastapi import FastAPI
from starlette.concurrency import run_in_threadpool

app = FastAPI()

def sync_operation():
    # Синхронная блокирующая операция
    return "синхронное значение"

@app.get("/")
async def root():
    result = await run_in_threadpool(sync_operation)
    return {"result": result}

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users")
def read_users(db = Depends(get_db)):
    return {"users": db.query(User).all()}

from fastapi import FastAPI, Depends, HTTPException, status
app = FastAPI()

async def verify_token(token: str):
    if token != 'secret-token':
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Неверный токен")
    return token

@app.get("/secure-data")
async def secure_data(token: str = Depends(verify_token)):
    return {'data': "Конфиденциальная информация"}