ابدأ بـ بايثون فاست إي بي آي: خطوتك الأولى العملية اليوم

# فلاسك
from flask import Flask
app = Flask(__name__)

@app.route("/items/<int:item_id>")
def get_item(item_id):
    return {"id": item_id}

# فاست إيه بي آي
from fastapi import FastAPI
app = FastAPI()

@app.get("/items/{item_id}")
def get_item(item_id: int):
    return {"id": item_id}
# + تحقق تلقائي، توثيق تلقائي، تلميحات الأنواع

# tests/test_posts.py
import pytest

def test_create_post_requires_auth(client):
    r = client.post("/posts", json={"title": "Test",
                                  "content": "Hello world"})
    assert r.status_code == 401

def test_create_post(client, auth_headers):
    r = client.post("/posts", headers=auth_headers, json={
        "title": "Mon premier post",
        "content": "Contenu interessant ici",
        "tags": ["python", "fastapi"],
        "published": True
    })
    assert r.status_code == 201
    data = r.json()
    assert data["slug"] == "mon-premier-post"
    assert len(data["tags"]) == 2

def test_list_posts_only_published(client, auth_headers):
    client.post("/posts", headers=auth_headers,
                json={"title": "Draft", "content": "...", "published": False})
    client.post("/posts", headers=auth_headers,
                json={"title": "Published", "content": "...", "published": True})
    
    r = client.get("/posts")
    assert len(r.json()["items"]) == 1

def test_only_author_can_delete(client):
    h1 = make_user_and_login(client, "alice", "alice@x.com")
    h2 = make_user_and_login(client, "bob", "bob@x.com")
    
    r = client.post("/posts", headers=h1,
                     json={"title": "Mien", "content": "...", "published": True})
    slug = r.json()["slug"]
    
    # يحاول Bob حذف منشور Alice
    r = client.delete(f"/posts/{slug}", headers=h2)
    assert r.status_code == 403
    
    # يمكن لـ Alice
    r = client.delete(f"/posts/{slug}", headers=h1)
    assert r.status_code == 204

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.db import Base, get_db

TestEngine = create_engine("sqlite:///./test.db",
                            connect_args={"check_same_thread": False})
TestSession = sessionmaker(bind=TestEngine)

def override_db():
    db = TestSession()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_db

@pytest.fixture(autouse=True)
def reset_db():
    Base.metadata.create_all(bind=TestEngine)
    yield
    Base.metadata.drop_all(bind=TestEngine)

@pytest.fixture
def client():
    return TestClient(app)

@pytest.fixture
def auth_headers(client):
    client.post("/auth/register", json={
        "email": "test@x.com", "username": "test",
        "password": "secret123"
    })
    r = client.post("/auth/login", data={
        "username": "test@x.com", "password": "secret123"
    })
    return {"Authorization": f'Bearer {r.json()["access_token"]}'}

FROM python:3.11-slim AS builder
WORKDIR /build
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

FROM python:3.11-slim
RUN useradd -m app
USER app
WORKDIR /home/app
ENV PATH=/home/app/.local/bin:$PATH

COPY --from=builder /root/.local /home/app/.local
COPY --chown=app:app . .

EXPOSE 8000
HEALTHCHECK CMD curl -fsS http://localhost:8000/health || exit 1

ENTRYPOINT ["./entrypoint.sh"]
CMD ["gunicorn", "-c", "gunicorn.conf.py", "app.main:app"]

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql://blog:pwd@db/blogdb
      SECRET_KEY: change-me-in-prod
      REFRESH_SECRET_KEY: change-me-too
    depends_on:
      db: {condition: service_healthy}
    restart: unless-stopped
  
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: blog
      POSTGRES_PASSWORD: pwd
      POSTGRES_DB: blogdb
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U blog"]
      interval: 5s
      retries: 5

volumes:
  pgdata:

# .github/workflows/ci.yml
name: CI
on:
  push:
    branches: [main]
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.11"}
      - run: pip install -r requirements.txt
      - run: pip install pytest pytest-cov httpx
      - run: pytest --cov=app --cov-report=xml
      - uses: codecov/codecov-action@v4
  
  build:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == "refs/heads/main"
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - uses: docker/build-push-action@v5
        with:
          push: true
          tags: ghcr.io/${{ github.repository }}:latest

# محلي كامل
docker-compose up -d
docker-compose logs -f api
docker-compose exec api alembic upgrade head

# زيارة
#   http://localhost:8000/docs
#   http://localhost:8000/redoc

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client deconnecte")

npm install -g wscat
wscat -c ws://localhost:8000/ws
> Hello
< Echo: Hello

class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []
    
    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)
    
    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)
    
    async def broadcast(self, message: str):
        for connection in self.active:
            try:
                await connection.send_text(message)
            except:
                pass

manager = ConnectionManager()

@app.websocket("/chat/{username}")
async def chat(ws: WebSocket, username: str):
    await manager.connect(ws)
    await manager.broadcast(f">> {username} a rejoint")
    
    try:
        while True:
            msg = await ws.receive_text()
            await manager.broadcast(f"[{username}] {msg}")
    except WebSocketDisconnect:
        manager.disconnect(ws)
        await manager.broadcast(f"<< {username} a quitte")

@app.websocket("/json")
async def json_ws(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_json()
            await ws.send_json({"echo": data, "server_time": str(datetime.now())})
    except WebSocketDisconnect:
        pass

from fastapi import WebSocketException, status

@app.websocket("/ws")
async def ws(ws: WebSocket, token: str):
    try:
        payload = decode_token(token)
        user_id = payload["sub"]
    except:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    
    await ws.accept()
    await ws.send_text(f"Connecte en tant que {user_id}")
    ...

# العميل: ws://localhost:8000/ws?token=eyJ...