Zamonaviy Web API'lar uchun To'liq Darslik
Python bilan yuqori samarali API'larni yaratish bo'yicha qo'llanma
FastAPI - bu Python 3.7+ uchun standart Python type hint'lariga asoslangan, zamonaviy, tez (yuqori samarali) web framework bo'lib, API'larni yaratish uchun mo'ljallangan.
pip install fastapi uvicorn[standard]
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello World"}
uvicorn main:app --reload
Path parametrlari URL yo'lidan qiymatlarni olish imkonini beradi.
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"item_id": item_id}
@app.get("/users/{user_id}/posts/{post_id}")
def read_user_post(user_id: int, post_id: int):
return {
"user_id": user_id,
"post_id": post_id
}
Query parametrlari URL'dagi ? belgisidan keyin paydo bo'ladigan kalit-qiymat juftliklaridir.
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
return {
"skip": skip,
"limit": limit
}
URL: http://localhost:8000/items/?skip=0&limit=20
from typing import Optional
@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
if q:
return {"item_id": item_id, "q": q}
return {"item_id": item_id}
Avtomatik tekshiruv bilan request body sxemalarini belgilash uchun Pydantic modellaridan foydalaning.
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
def create_item(item: Item):
return {
"name": item.name,
"price": item.price
}
API javoblarining strukturasini belgilang va avtomatik hujjatlar yarating.
class ItemResponse(BaseModel):
id: int
name: str
price: float
@app.post("/items/", response_model=ItemResponse)
def create_item(item: Item):
# Imagine this saves to database
return {
"id": 1,
"name": item.name,
"price": item.price
}
Turli vaziyatlar uchun mos status kodlarini qaytaring.
from fastapi import status
@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
return item
status.HTTP_200_OK # Muvaffaqiyat status.HTTP_201_CREATED # Resurs yaratildi status.HTTP_204_NO_CONTENT # Muvaffaqiyat, kontent yo'q status.HTTP_400_BAD_REQUEST # Noto'g'ri so'rov status.HTTP_401_UNAUTHORIZED # Autentifikatsiya qilinmagan status.HTTP_403_FORBIDDEN # Ruxsat berilmagan status.HTTP_404_NOT_FOUND # Resurs topilmadi status.HTTP_500_INTERNAL_SERVER_ERROR
HTTPException bilan xatolarni chiroyli boshqaring.
from fastapi import HTTPException
items = {"1": "Item One"}
@app.get("/items/{item_id}")
def read_item(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=404,
detail="Item not found"
)
return {"item": items[item_id]}
from fastapi.responses import JSONResponse
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
return JSONResponse(
status_code=400,
content={"message": str(exc)}
)
Bir nechta endpoint'lar bo'ylab umumiy mantiqni qayta ishlating.
from fastapi import Depends
def get_query_param(q: str = None):
return q
@app.get("/items/")
def read_items(query: str = Depends(get_query_param)):
return {"query": query}
def get_db():
db = Database()
try:
yield db
finally:
db.close()
@app.get("/users/")
def read_users(db = Depends(get_db)):
return db.get_all_users()
Har bir so'rovdan oldin yoki keyin ishlaydigan maxsus mantiq qo'shing.
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
import time
@app.middleware("http")
async def add_process_time(request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
Javob qaytarilgandan keyin vazifalarni fon rejimida bajarish.
from fastapi import BackgroundTasks
def send_email(email: str, message: str):
# Simulate sending email
print(f"Sending email to {email}: {message}")
@app.post("/send-notification/")
def send_notification(
email: str,
background_tasks: BackgroundTasks
):
background_tasks.add_task(
send_email,
email,
"Thanks for subscribing!"
)
return {"message": "Notification sent"}
FastAPI bilan fayl yuklashni oson boshqaring.
from fastapi import File, UploadFile
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
contents = await file.read()
return {
"filename": file.filename,
"size": len(contents)
}
from typing import List
@app.post("/uploadfiles/")
async def upload_files(files: List[UploadFile] = File(...)):
return {
"filenames": [file.filename for file in files]
}
API'ingizda xavfsiz autentifikatsiyani amalga oshiring.
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/users/me")
def read_users_me(token: str = Depends(oauth2_scheme)):
return {"token": token}
from datetime import datetime, timedelta
import jwt
SECRET_KEY = "your-secret-key"
def create_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=30)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")
FastAPI avtomatik ravishda interaktiv API hujjatlarini yaratadi.
http://127.0.0.1:8000/docshttp://127.0.0.1:8000/redochttp://127.0.0.1:8000/openapi.jsonapp = FastAPI(
title="My API",
description="A comprehensive API",
version="1.0.0",
docs_url="/documentation",
redoc_url="/redocumentation"
)
def process_item(item_id: int, name: str) -> dict:
return {"id": item_id, "name": name}
from fastapi import APIRouter
router = APIRouter(prefix="/items", tags=["items"])
@router.get("/")
def read_items():
return []
app.include_router(router)
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
secret_key: str
class Config:
env_file = ".env"
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" engine = create_engine(SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/")
def read_users(db: Session = Depends(get_db)):
return db.query(User).all()
from fastapi.testclient import TestClient
client = TestClient(app)
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test", "price": 10.5}
)
assert response.status_code == 200
assert response.json()["name"] == "Test"
pip install pytest
pip install gunicorn gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker
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", "8000"]
# .env file DATABASE_URL=postgresql://user:pass@localhost/db SECRET_KEY=your-secret-key-here DEBUG=False