from sqlalchemy.ext.asyncio import create_async_engine

class TenantDBRouter:
    """Har tenant uchun alohida connection"""
    def __init__(self):
        self._engines: dict[str, AsyncEngine] = {}
    
    async def get_engine(self, tenant_id: str) -> AsyncEngine:
        if tenant_id not in self._engines:
            # Metadata DB'dan tenant connection string olish
            conn_str = await self._resolve_tenant_db(tenant_id)
            self._engines[tenant_id] = create_async_engine(
                conn_str, pool_size=5, max_overflow=10
            )
        return self._engines[tenant_id]
    
    async def _resolve_tenant_db(self, tenant_id: str) -> str:
        # metadata DB'dan lookup
        tenant = await metadata_db.fetch_one(
            "SELECT db_host, db_name FROM tenants WHERE id = $1", tenant_id
        )
        return f"postgresql+asyncpg://{tenant.db_host}/{tenant.db_name}"

-- Har tenant uchun schema
CREATE SCHEMA tenant_acme;
CREATE SCHEMA tenant_globex;

-- Har schema'da jadvallar
CREATE TABLE tenant_acme.orders (id UUID PRIMARY KEY, amount DECIMAL);
CREATE TABLE tenant_globex.orders (id UUID PRIMARY KEY, amount DECIMAL);

-- Query vaqtida search_path
SET search_path TO tenant_acme;
SELECT * FROM orders;  -- avtomatik tenant_acme.orders

CREATE TABLE orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    user_id UUID NOT NULL,
    amount DECIMAL(10,2) NOT NULL,
    status TEXT NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    -- MUHIM: har composite index tenant_id dan boshlanadi
    CONSTRAINT fk_tenant FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);

-- tenant_id doimo birinchi!
CREATE INDEX idx_orders_tenant_status 
    ON orders(tenant_id, status) 
    WHERE status != 'archived';

CREATE INDEX idx_orders_tenant_user_created 
    ON orders(tenant_id, user_id, created_at DESC);

-- RLS — oxirgi himoya qatlami
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON orders
    USING (tenant_id = current_setting('app.tenant_id')::uuid);

from prometheus_client import Histogram
import time

# Histogram — distribution'ni saqlaydi, percentile hisoblash uchun
http_duration = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint', 'status'],
    # bucket'lar — bu oraliqlarda counter bo'ladi
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

@app.middleware("http")
async def track_latency(request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    
    http_duration.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).observe(duration)
    
    return response

# p99 oxirgi 5 daqiqa ichida
histogram_quantile(0.99, 
  sum(rate(http_request_duration_seconds_bucket[5m])) by (le)
)

# Per-endpoint p95
histogram_quantile(0.95,
  sum(rate(http_request_duration_seconds_bucket[5m])) by (le, endpoint)
)

# RPS
sum(rate(http_request_duration_seconds_count[1m]))

TOKEN_BUCKET_LUA = """
local key = KEYS[1]
local rate = tonumber(ARGV[1])         -- tokens per second
local capacity = tonumber(ARGV[2])     -- bucket size
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

local bucket = redis.call('HMGET', key, 'tokens', 'last')
local tokens = tonumber(bucket[1]) or capacity
local last = tonumber(bucket[2]) or now

-- tokens to'ldirish
local elapsed = math.max(0, now - last)
tokens = math.min(capacity, tokens + elapsed * rate)

if tokens < requested then
    return {0, tokens}  -- rad etish
end

tokens = tokens - requested
redis.call('HMSET', key, 'tokens', tokens, 'last', now)
redis.call('EXPIRE', key, 3600)
return {1, tokens}
"""

class TenantRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.script = redis_client.register_script(TOKEN_BUCKET_LUA)
    
    async def check(self, tenant_id: str, cost: int = 1) -> tuple[bool, int]:
        # Tenant tier'iga qarab limit
        limits = TIER_LIMITS[tenant_tier(tenant_id)]
        
        allowed, remaining = await self.script(
            keys=[f"rl:{tenant_id}"],
            args=[limits.rate, limits.burst, time.time(), cost]
        )
        return bool(allowed), int(remaining)

TIER_LIMITS = {
    "free":       Limits(rate=10,  burst=20),    # 10 req/sec
    "pro":        Limits(rate=100, burst=200),
    "enterprise": Limits(rate=1000, burst=2000),
}

from celery import Celery

app = Celery('myapp', broker='redis://localhost:6379/0')

app.conf.task_routes = {
    'app.tasks.send_email':    {'queue': 'fast'},
    'app.tasks.process_order': {'queue': 'fast'},
    'app.tasks.export_report': {'queue': 'heavy'},
    'app.tasks.ml_inference':  {'queue': 'gpu'},
}

# Per-tenant rate limit
@app.task(bind=True, rate_limit='100/m')
def send_email(self, tenant_id, to, subject):
    ...

# Og'ir vazifalar
@app.task(bind=True, rate_limit='5/m', soft_time_limit=600)
def export_report(self, tenant_id, report_id):
    ...

services:
  worker-fast:
    image: myapp
    command: celery -A app worker -Q fast --concurrency=20 --max-tasks-per-child=1000
    deploy:
      replicas: 3

  worker-heavy:
    image: myapp
    command: celery -A app worker -Q heavy --concurrency=4 --max-tasks-per-child=100
    deploy:
      replicas: 2
      resources:
        limits:
          memory: 4G
          cpus: '2.0'

  worker-gpu:
    image: myapp-ml
    command: celery -A app worker -Q gpu --concurrency=1
    deploy:
      resources:
        reservations:
          devices:
            - capabilities: [gpu]

from fastapi import Response, Cookie, HTTPException
import secrets

SESSION_TTL = 3600 * 24 * 7  # 7 kun

async def login(email: str, password: str, response: Response):
    user = await verify_credentials(email, password)
    if not user:
        raise HTTPException(401, "Invalid credentials")
    
    # Session ID yaratish
    sid = secrets.token_urlsafe(32)
    await redis.setex(
        f"session:{sid}",
        SESSION_TTL,
        json.dumps({
            "user_id": str(user.id),
            "tenant_id": str(user.tenant_id),
            "created_at": time.time(),
            "ip": request.client.host,
            "user_agent": request.headers.get("user-agent", "")
        })
    )
    
    # Secure cookie
    response.set_cookie(
        key="sid",
        value=sid,
        max_age=SESSION_TTL,
        httponly=True,    # JS'dan o'qib bo'lmaydi (XSS himoya)
        secure=True,      # faqat HTTPS
        samesite="lax",   # CSRF himoya
    )
    return {"status": "logged_in"}

async def get_current_user(sid: str = Cookie(None)):
    if not sid:
        raise HTTPException(401)
    data = await redis.get(f"session:{sid}")
    if not data:
        raise HTTPException(401)
    return User(**json.loads(data))

async def logout(sid: str = Cookie(None)):
    await redis.delete(f"session:{sid}")

"""Production-grade hybrid: short JWT + long refresh token"""
import jwt
from datetime import datetime, timedelta, timezone

ACCESS_TOKEN_TTL = timedelta(minutes=15)
REFRESH_TOKEN_TTL = timedelta(days=30)

class TokenService:
    def __init__(self, secret: str, algorithm: str = "HS256"):
        self.secret = secret
        self.algorithm = algorithm
    
    def create_access(self, user: User) -> str:
        now = datetime.now(timezone.utc)
        payload = {
            "sub": str(user.id),
            "tenant": str(user.tenant_id),
            "roles": user.roles,
            "iat": now,
            "exp": now + ACCESS_TOKEN_TTL,
            "type": "access",
        }
        return jwt.encode(payload, self.secret, algorithm=self.algorithm)
    
    async def create_refresh(self, user: User) -> str:
        token = secrets.token_urlsafe(32)
        # Redis'da saqlaymiz — revoke qilish uchun
        await redis.setex(
            f"refresh:{token}",
            REFRESH_TOKEN_TTL.total_seconds(),
            json.dumps({"user_id": str(user.id), "created_at": time.time()})
        )
        return token
    
    def verify_access(self, token: str) -> dict:
        try:
            payload = jwt.decode(token, self.secret, algorithms=[self.algorithm])
            if payload.get("type") != "access":
                raise HTTPException(401, "Invalid token type")
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(401, "Token expired")
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token")
    
    async def refresh(self, refresh_token: str) -> dict:
        data = await redis.get(f"refresh:{refresh_token}")
        if not data:
            raise HTTPException(401, "Invalid refresh token")
        
        info = json.loads(data)
        user = await get_user(info["user_id"])
        
        # Rotation — eski refresh tokenni o'chirib, yangisini beramiz
        await redis.delete(f"refresh:{refresh_token}")
        
        return {
            "access_token": self.create_access(user),
            "refresh_token": await self.create_refresh(user),
        }

from authlib.integrations.starlette_client import OAuth

oauth = OAuth()
oauth.register(
    name='google',
    client_id=os.getenv('GOOGLE_CLIENT_ID'),
    client_secret=os.getenv('GOOGLE_CLIENT_SECRET'),
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

@app.get('/auth/google')
async def google_login(request: Request):
    redirect_uri = request.url_for('google_callback')
    return await oauth.google.authorize_redirect(request, redirect_uri)

@app.get('/auth/google/callback')
async def google_callback(request: Request):
    token = await oauth.google.authorize_access_token(request)
    user_info = token.get('userinfo') or await oauth.google.parse_id_token(request, token)
    
    # Find or create user
    user = await find_or_create_user(
        email=user_info['email'],
        name=user_info.get('name'),
        google_sub=user_info['sub'],
    )
    
    # Issue your own session/tokens
    return issue_tokens(user)

from webauthn import generate_registration_options, verify_registration_response
from webauthn.helpers.structs import AuthenticatorSelectionCriteria, UserVerificationRequirement

RP_ID = "myapp.com"
RP_NAME = "My App"

@app.post("/webauthn/register/begin")
async def register_begin(user: User = Depends(get_current_user)):
    options = generate_registration_options(
        rp_id=RP_ID,
        rp_name=RP_NAME,
        user_id=str(user.id).encode(),
        user_name=user.email,
        user_display_name=user.name,
        authenticator_selection=AuthenticatorSelectionCriteria(
            user_verification=UserVerificationRequirement.REQUIRED,
        ),
    )
    # Challenge'ni session'da saqlaymiz
    await redis.setex(f"webauthn:challenge:{user.id}", 300, options.challenge)
    return options

@app.post("/webauthn/register/complete")
async def register_complete(credential: dict, user: User = Depends(get_current_user)):
    expected_challenge = await redis.get(f"webauthn:challenge:{user.id}")
    verification = verify_registration_response(
        credential=credential,
        expected_challenge=expected_challenge,
        expected_origin=f"https://{RP_ID}",
        expected_rp_id=RP_ID,
    )
    # Public key'ni DB'ga saqlaymiz
    await db.save_credential(
        user_id=user.id,
        credential_id=verification.credential_id,
        public_key=verification.credential_public_key,
        sign_count=verification.sign_count,
    )
    return {"ok": True}

CREATE TABLE roles (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    name TEXT NOT NULL,
    UNIQUE (tenant_id, name)
);

CREATE TABLE permissions (
    id UUID PRIMARY KEY,
    resource TEXT NOT NULL,    -- 'orders', 'reports'
    action TEXT NOT NULL,      -- 'read', 'write', 'delete'
    UNIQUE (resource, action)
);

CREATE TABLE role_permissions (
    role_id UUID REFERENCES roles(id) ON DELETE CASCADE,
    permission_id UUID REFERENCES permissions(id) ON DELETE CASCADE,
    PRIMARY KEY (role_id, permission_id)
);

CREATE TABLE user_roles (
    user_id UUID REFERENCES users(id) ON DELETE CASCADE,
    role_id UUID REFERENCES roles(id) ON DELETE CASCADE,
    PRIMARY KEY (user_id, role_id)
);

def can_access_report(user, report, context):
    """Report ko'rish qoidalari — murakkab"""
    # Admin hamma narsani
    if "admin" in user.roles:
        return True
    
    # Own department
    if user.department_id == report.department_id:
        return True
    
    # Manager — qo'l ostidagilar report'lari
    if user.role == "manager" and report.author_id in user.subordinate_ids:
        return True
    
    # Finance team — faqat ish vaqti
    if "finance" in user.roles and report.type == "financial":
        now = context["now"]
        if 9 <= now.hour < 18 and now.weekday() < 5:
            return True
    
    # Region restriction
    if report.classification == "confidential":
        if user.region != report.region:
            return False
    
    return False

# OpenFGA — Zanzibar'ning open-source versiyasi
from openfga_sdk import OpenFgaClient

fga = OpenFgaClient(...)

# Relationship yaratish
await fga.write(tuples=[{
    "user": "user:alice",
    "relation": "owner",
    "object": "document:report_2026"
}, {
    "user": "group:finance#member",
    "relation": "viewer",
    "object": "document:report_2026"
}])

# Permission check
result = await fga.check({
    "user": "user:carol",
    "relation": "can_view",
    "object": "document:report_2026"
})
# result.allowed = True (carol ∈ finance, finance ∋ viewer)

package app.authz

default allow = false

# Admin hamma narsa
allow {
    input.user.roles[_] == "admin"
}

# Owner o'zining resursi
allow {
    input.resource.owner_id == input.user.id
}

# Reader faqat publish'lar
allow {
    input.action == "read"
    input.resource.status == "published"
    input.user.roles[_] == "reader"
}

# Workday restriction finance uchun
allow {
    input.user.department == "finance"
    input.resource.type == "financial_report"
    workday
    time.clock(time.now_ns())[0] >= 9
    time.clock(time.now_ns())[0] < 18
}

workday {
    day := time.weekday(time.now_ns())
    day != "Saturday"
    day != "Sunday"
}

@app.get("/orders")
async def list_orders(page: int = 1, per_page: int = 20):
    offset = (page - 1) * per_page
    orders = await db.fetch_all(
        "SELECT * FROM orders WHERE tenant_id = $1 "
        "ORDER BY created_at DESC LIMIT $2 OFFSET $3",
        tenant_id, per_page, offset
    )
    total = await db.fetch_val("SELECT COUNT(*) FROM orders WHERE tenant_id = $1", tenant_id)
    return {
        "data": orders,
        "meta": {"page": page, "per_page": per_page, "total": total}
    }

import base64

def encode_cursor(value: dict) -> str:
    return base64.urlsafe_b64encode(json.dumps(value).encode()).decode()

def decode_cursor(cursor: str) -> dict:
    return json.loads(base64.urlsafe_b64decode(cursor.encode()))

@app.get("/orders")
async def list_orders(cursor: str | None = None, limit: int = 20):
    limit = min(limit, 100)
    
    if cursor:
        c = decode_cursor(cursor)
        rows = await db.fetch_all("""
            SELECT * FROM orders 
            WHERE tenant_id = $1 AND (created_at, id) < ($2, $3)
            ORDER BY created_at DESC, id DESC
            LIMIT $4
        """, tenant_id, c["created_at"], c["id"], limit + 1)
    else:
        rows = await db.fetch_all("""
            SELECT * FROM orders WHERE tenant_id = $1
            ORDER BY created_at DESC, id DESC LIMIT $2
        """, tenant_id, limit + 1)
    
    has_more = len(rows) > limit
    rows = rows[:limit]
    
    next_cursor = None
    if has_more and rows:
        last = rows[-1]
        next_cursor = encode_cursor({"created_at": last.created_at.isoformat(), "id": str(last.id)})
    
    return {"data": rows, "next_cursor": next_cursor, "has_more": has_more}

from fastapi import HTTPException
from fastapi.responses import JSONResponse

class APIError(HTTPException):
    def __init__(self, status: int, type_: str, title: str, detail: str = "", **extensions):
        self.type = type_
        self.title = title
        self.detail = detail
        self.extensions = extensions
        super().__init__(status_code=status, detail=detail)

@app.exception_handler(APIError)
async def api_error_handler(request, exc: APIError):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "type": exc.type,
            "title": exc.title,
            "status": exc.status_code,
            "detail": exc.detail,
            "instance": str(request.url),
            **exc.extensions,
        },
        media_type="application/problem+json"
    )

# Ishlatish
raise APIError(
    status=422,
    type_="https://api.myapp.com/errors/validation",
    title="Validation failed",
    detail="amount must be positive",
    field="amount",
    received=-100,
)

{
  "type": "https://api.myapp.com/errors/validation",
  "title": "Validation failed",
  "status": 422,
  "detail": "amount must be positive",
  "instance": "https://api.myapp.com/orders",
  "field": "amount",
  "received": -100
}

query GetUserDashboard($userId: ID!) {
  user(id: $userId) {
    name
    email
    orders(limit: 5, status: ACTIVE) {
      id
      total
      items {
        product { name, price }
        quantity
      }
    }
    notifications(unread: true) {
      id
      message
    }
  }
}

syntax = "proto3";

package orders.v1;

service OrderService {
  rpc CreateOrder(CreateOrderRequest) returns (Order);
  rpc GetOrder(GetOrderRequest) returns (Order);
  rpc ListOrders(ListOrdersRequest) returns (ListOrdersResponse);
  rpc StreamOrderUpdates(StreamRequest) returns (stream OrderUpdate);
}

message Order {
  string id = 1;
  string tenant_id = 2;
  string user_id = 3;
  repeated OrderItem items = 4;
  double total = 5;
  OrderStatus status = 6;
  google.protobuf.Timestamp created_at = 7;
}

enum OrderStatus {
  PENDING = 0;
  PAID = 1;
  SHIPPED = 2;
  DELIVERED = 3;
  CANCELLED = 4;
}

import grpc
from orders_pb2_grpc import OrderServiceServicer, add_OrderServiceServicer_to_server
from orders_pb2 import Order, OrderStatus

class OrderServicer(OrderServiceServicer):
    async def CreateOrder(self, request, context):
        order = await db.create_order(request.user_id, request.items)
        return Order(
            id=str(order.id),
            tenant_id=order.tenant_id,
            user_id=order.user_id,
            total=order.total,
            status=OrderStatus.PENDING,
        )
    
    async def StreamOrderUpdates(self, request, context):
        """Server streaming — live updates"""
        async for update in order_update_stream(request.user_id):
            yield OrderUpdate(order_id=update.id, status=update.status)

async def serve():
    server = grpc.aio.server()
    add_OrderServiceServicer_to_server(OrderServicer(), server)
    server.add_insecure_port('[::]:50051')
    await server.start()
    await server.wait_for_termination()

import { initTRPC } from '@trpc/server';
import { z } from 'zod';

const t = initTRPC.create();

export const appRouter = t.router({
  getOrder: t.procedure
    .input(z.object({ id: z.string() }))
    .query(async ({ input }) => {
      return await db.order.findUnique({ where: { id: input.id } });
    }),
  
  createOrder: t.procedure
    .input(z.object({
      userId: z.string(),
      items: z.array(z.object({ productId: z.string(), qty: z.number() }))
    }))
    .mutation(async ({ input }) => {
      return await db.order.create({ data: input });
    }),
});

export type AppRouter = typeof appRouter;

// Client — backend type'larini to'liq biladi
const { data, isLoading } = trpc.getOrder.useQuery({ id: "abc-123" });
//      ^^^^ type: Order | undefined, fully typed

const createOrder = trpc.createOrder.useMutation();
createOrder.mutate({ userId: "u-1", items: [{ productId: "p-1", qty: 2 }] });
// If you pass wrong type — TypeScript error at compile time

from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
import time, uuid, logging

app = FastAPI()

# CORS — eng avval
app.add_middleware(CORSMiddleware,
    allow_origins=["https://app.mysite.com"],
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    max_age=3600,  # preflight cache
)

@app.middleware("http")
async def request_lifecycle(request: Request, call_next):
    """Request-to-response hayot siklini boshqarish"""
    # ── 1. REQUEST ID (tracing) ──
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request.state.request_id = request_id

    # ── 2. START TIMING ──
    start = time.perf_counter()

    # ── 3. STRUCTURED LOGGING ──
    logger = logging.getLogger("request")
    logger.info({
        "event": "request_start",
        "request_id": request_id,
        "method": request.method,
        "path": request.url.path,
        "ip": request.client.host,
    })

    try:
        # ── 4. PROCESS (next middleware / handler) ──
        response: Response = await call_next(request)

        # ── 5. ADD RESPONSE HEADERS ──
        duration_ms = (time.perf_counter() - start) * 1000
        response.headers["X-Request-ID"] = request_id
        response.headers["X-Response-Time"] = f"{duration_ms:.2f}ms"
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"

        # ── 6. LOG COMPLETION ──
        logger.info({
            "event": "request_end",
            "request_id": request_id,
            "status": response.status_code,
            "duration_ms": round(duration_ms, 2),
        })

        # ── 7. SLOW REQUEST ALERT ──
        if duration_ms > 1000:
            logger.warning({
                "event": "slow_request",
                "request_id": request_id,
                "path": request.url.path,
                "duration_ms": round(duration_ms, 2),
            })

        return response

    except Exception as exc:
        duration_ms = (time.perf_counter() - start) * 1000
        logger.error({
            "event": "request_error",
            "request_id": request_id,
            "error": str(exc),
            "duration_ms": round(duration_ms, 2),
        })
        raise

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

# RFC 9457 — Problem Details format
class ProblemDetail(BaseModel):
    type: str            # URI — mashinaga o'qiladi
    title: str           # Qisqa xato nomi
    status: int          # HTTP status code
    detail: str          # Aniq, foydali ma'lumot
    instance: str = ""   # Qaysi URL'da bo'ldi

# Custom exception
class AppError(Exception):
    def __init__(self, type: str, title: str, detail: str, status: int = 400):
        self.type = type
        self.title = title
        self.detail = detail
        self.status = status

app = FastAPI()

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=exc.status,
        content={
            "type": f"https://docs.myapp.com/errors/{exc.type}",
            "title": exc.title,
            "status": exc.status,
            "detail": exc.detail,
            "instance": str(request.url),
        },
        headers={"Content-Type": "application/problem+json"}
    )

# Ishlatish:
@app.post("/orders")
async def create_order(data: OrderCreate):
    if data.amount <= 0:
        raise AppError(
            type="invalid-amount",
            title="Invalid Order Amount",
            detail=f"Amount must be positive, got {data.amount}",
            status=400
        )
    if not await check_inventory(data.product_id):
        raise AppError(
            type="out-of-stock",
            title="Product Out of Stock",
            detail=f"Product {data.product_id} is currently unavailable",
            status=409
        )
    ...

# Response:
# HTTP 409
# Content-Type: application/problem+json
# {
#   "type": "https://docs.myapp.com/errors/out-of-stock",
#   "title": "Product Out of Stock",
#   "status": 409,
#   "detail": "Product abc-123 is currently unavailable",
#   "instance": "/orders"
# }

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from anthropic import AsyncAnthropic
import asyncio, json

app = FastAPI()
client = AsyncAnthropic()

@app.post("/api/chat/stream")
async def chat_stream(prompt: str, user_id: str):
    """LLM javobini SSE orqali streaming"""

    async def generate():
        # SSE format: "data: {json}\n\n"
        # Connection o'rnatildi signali
        yield f"data: {json.dumps({'type': 'start'})}\n\n"

        full_text = ""
        try:
            # Claude streaming API
            async with client.messages.stream(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}]
            ) as stream:
                async for text_chunk in stream.text_stream:
                    full_text += text_chunk
                    # Har token'ni darhol yuborish
                    yield f"data: {json.dumps({'type': 'chunk', 'text': text_chunk})}\n\n"

            # Tugash signali
            yield f"data: {json.dumps({'type': 'done', 'total_chars': len(full_text)})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",   # Nginx buffering'ni o'chirish
            "Connection": "keep-alive",
        }
    )

# Frontend (JavaScript) tomonida:
# const es = new EventSource('/api/chat/stream?prompt=...');
# es.onmessage = (e) => {
#   const data = JSON.parse(e.data);
#   if (data.type === 'chunk') appendText(data.text);
#   if (data.type === 'done') es.close();
# };

async def update_user(user_id: str, data: dict):
    # 1. DB yangilanadi
    user = await db.update(user_id, data)
    # 2. Cache yangilanadi bir vaqtda
    await redis.setex(f"user:{user_id}", 3600, json.dumps(user.dict()))
    return user

async def increment_view_count(post_id: str):
    # Cache'ga tez yozamiz
    await redis.incr(f"views:{post_id}")
    # DB'ga keyingi batch'da yoziladi (har 30 sekundda)

# Background job
async def flush_view_counts():
    while True:
        keys = await redis.keys("views:*")
        if keys:
            async with db.transaction():
                for key in keys:
                    post_id = key.split(":")[1]
                    count = int(await redis.get(key))
                    await db.execute(
                        "UPDATE posts SET view_count = view_count + $1 WHERE id = $2",
                        count, post_id
                    )
                    await redis.delete(key)
        await asyncio.sleep(30)

async def get_product(product_id: str) -> dict:
    # Global version prefix — mass invalidate uchun
    version = await redis.get("cache:products:version") or "1"
    key = f"product:v{version}:{product_id}"
    
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)
    
    product = await db.fetch_product(product_id)
    await redis.setex(key, 3600, json.dumps(product.dict()))
    return product.dict()

async def invalidate_all_products():
    """Bitta product schema o'zgardi → hammasini bekor qilish"""
    await redis.incr("cache:products:version")
    # Eski key'lar TTL bilan tabiiy o'ladi (disk cleanup)

async def get_with_stampede_protection(key: str, fetch_fn, ttl: int = 300):
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)
    
    # Distributed lock — faqat bitta request DB'ga boradi
    lock_key = f"lock:{key}"
    acquired = await redis.set(lock_key, "1", nx=True, ex=10)
    
    if not acquired:
        # Boshqa request DB'dan olmoqda — kutamiz
        for _ in range(20):
            await asyncio.sleep(0.1)
            cached = await redis.get(key)
            if cached:
                return json.loads(cached)
        # 2 sek kutdik, hali yo'q — DB'ga o'zimiz boramiz
    
    try:
        value = await fetch_fn()
        await redis.setex(key, ttl, json.dumps(value))
        return value
    finally:
        await redis.delete(lock_key)

-- Composite — LEFT-MOST RULE muhim
CREATE INDEX idx_orders_tenant_user_date 
  ON orders (tenant_id, user_id, created_at DESC);

-- Ishlaydi:
SELECT * FROM orders WHERE tenant_id = ? AND user_id = ? ORDER BY created_at DESC;
SELECT * FROM orders WHERE tenant_id = ? AND user_id = ?;
SELECT * FROM orders WHERE tenant_id = ?;

-- Ishlamaydi (index skip):
SELECT * FROM orders WHERE user_id = ?;              -- tenant_id yo'q
SELECT * FROM orders WHERE created_at > ?;           -- chap 2 yo'q


-- Partial index — faqat kerakli subsetga
CREATE INDEX idx_orders_active 
  ON orders (tenant_id, created_at DESC)
  WHERE status IN ('pending', 'processing');
-- 90% order'lar "delivered" bo'lsa, index 10x kichikroq


-- Covering index (INCLUDE) — Index Only Scan
CREATE INDEX idx_users_email_covering 
  ON users (email) 
  INCLUDE (name, avatar_url);
-- SELECT name, avatar_url FROM users WHERE email = '...' 
-- → index'dan to'g'ridan to'g'ri, heap'ga bormaydi


-- Expression index
CREATE INDEX idx_users_lower_email ON users (lower(email));
-- SELECT * FROM users WHERE lower(email) = '...' uchun


-- UNIQUE partial — soft delete uchun
CREATE UNIQUE INDEX idx_users_email_active 
  ON users (email) 
  WHERE deleted_at IS NULL;
-- Soft-deleted user'ning email'ini qayta ishlatish mumkin

-- JSONB — flexible schema
CREATE TABLE events (
    id UUID PRIMARY KEY,
    tenant_id UUID NOT NULL,
    event_type TEXT,
    properties JSONB
);

-- GIN — har JSONB path'ga index
CREATE INDEX idx_events_properties ON events USING GIN (properties);

-- Tez query'lar:
SELECT * FROM events WHERE properties @> '{"user_type": "premium"}';
SELECT * FROM events WHERE properties ? 'campaign_id';
SELECT * FROM events WHERE properties -> 'metadata' ->> 'source' = 'google';

-- JSONB path ops class (tez va kichik, lekin kam ops)
CREATE INDEX idx_events_props_path ON events USING GIN (properties jsonb_path_ops);


-- Array
CREATE TABLE articles (id SERIAL, title TEXT, tags TEXT[]);
CREATE INDEX idx_articles_tags ON articles USING GIN (tags);

SELECT * FROM articles WHERE tags @> ARRAY['python', 'web'];   -- hamma
SELECT * FROM articles WHERE tags && ARRAY['python', 'rust'];  -- kamida 1


-- Full-text search
ALTER TABLE articles ADD COLUMN search_vector tsvector
  GENERATED ALWAYS AS (
    setweight(to_tsvector('english', coalesce(title, '')), 'A') ||
    setweight(to_tsvector('english', coalesce(body, '')), 'B')
  ) STORED;

CREATE INDEX idx_articles_search ON articles USING GIN (search_vector);

SELECT *, ts_rank(search_vector, query) AS rank
FROM articles, plainto_tsquery('english', 'backend architecture') query
WHERE search_vector @@ query
ORDER BY rank DESC;

CREATE TABLE metrics (
    id BIGSERIAL,
    tenant_id UUID,
    metric_name TEXT,
    value DOUBLE PRECISION,
    ts TIMESTAMPTZ NOT NULL
);

-- Time asosida BRIN (time-series natural ordering)
CREATE INDEX idx_metrics_ts_brin 
  ON metrics USING BRIN (ts) 
  WITH (pages_per_range = 128);

-- tenant_id composite BRIN
CREATE INDEX idx_metrics_tenant_ts_brin 
  ON metrics USING BRIN (tenant_id, ts);

-- Ishlatilmayotgan index'lar
SELECT 
    schemaname, tablename, indexname,
    pg_size_pretty(pg_relation_size(indexrelid)) AS size,
    idx_scan
FROM pg_stat_user_indexes
WHERE idx_scan < 50  -- juda kam ishlatilgan
  AND NOT indisunique  -- unique emas (unique'ni tashlab bo'lmaydi)
ORDER BY pg_relation_size(indexrelid) DESC
LIMIT 20;

-- Dublikat index'lar
SELECT indrelid::regclass AS table, 
       array_agg(indexrelid::regclass) AS indexes
FROM pg_index
GROUP BY indrelid, indkey
HAVING count(*) > 1;

EXPLAIN (ANALYZE, BUFFERS, VERBOSE, FORMAT TEXT)
SELECT o.*, u.name 
FROM orders o
JOIN users u ON u.id = o.user_id
WHERE o.tenant_id = 'xxx'
  AND o.status = 'pending'
ORDER BY o.created_at DESC
LIMIT 20;

/* Output (labels explained):
Limit  (cost=0.85..12.34 rows=20 width=128)
                    ^     ^       ^
         estimate start  total  rows
  (actual time=0.123..2.456 rows=20 loops=1)
                ^          ^
          actual start   actual end
  ->  Nested Loop  (cost=0.85..5234.12 rows=1234 width=128)
        ->  Index Scan using idx_orders_tenant_status on orders o
              (cost=0.43..234.56 rows=1234 width=64)
              (actual time=0.012..0.456 rows=1234 loops=1)
              Index Cond: (tenant_id = 'xxx' AND status = 'pending')
              Buffers: shared hit=45 read=3
                                ^^^    ^^^^
                               cache   disk
        ->  Index Scan using users_pkey on users u
              (cost=0.29..4.05 rows=1 width=64)
              (actual time=0.001..0.001 rows=1 loops=1234)
                                                 ^^^^^^^^^
                                              1234 ta iteration!
              Index Cond: (id = o.user_id)
Planning Time: 0.234 ms
Execution Time: 2.678 ms
*/

-- BOSHLANG'ICH (12 sekund!)
SELECT 
    u.name, COUNT(o.id) AS orders, SUM(o.amount) AS revenue,
    (SELECT MAX(created_at) FROM orders WHERE user_id = u.id) AS last_order
FROM users u
LEFT JOIN orders o ON o.user_id = u.id
WHERE u.tenant_id = 'xxx'
  AND u.created_at > NOW() - INTERVAL '1 year'
GROUP BY u.id, u.name
ORDER BY revenue DESC NULLS LAST
LIMIT 100;

-- EXPLAIN ANALYZE ko'rsatdi:
-- 1. Seq Scan on users (tenant_id + created_at index yo'q)
-- 2. Correlated subquery MAX(created_at) - har user uchun alohida query!
-- 3. Nested Loop 500k iteration

-- YECHIM:
-- Index
CREATE INDEX idx_users_tenant_created ON users (tenant_id, created_at DESC);
CREATE INDEX idx_orders_user_created ON orders (user_id, created_at DESC);

-- Query rewrite — correlated subquery'ni LATERAL bilan
SELECT 
    u.name, 
    COALESCE(stats.orders, 0) AS orders,
    COALESCE(stats.revenue, 0) AS revenue,
    stats.last_order
FROM users u
LEFT JOIN LATERAL (
    SELECT 
        COUNT(*) AS orders,
        SUM(amount) AS revenue,
        MAX(created_at) AS last_order
    FROM orders o 
    WHERE o.user_id = u.id
) stats ON true
WHERE u.tenant_id = 'xxx'
  AND u.created_at > NOW() - INTERVAL '1 year'
ORDER BY stats.revenue DESC NULLS LAST
LIMIT 100;

-- Natija: 12s → 80ms. 150x tezroq.

-- Extension
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;

-- Eng sekin query'lar
SELECT 
    substring(query, 1, 80) AS query,
    calls,
    round(total_exec_time::numeric, 2) AS total_ms,
    round(mean_exec_time::numeric, 2) AS avg_ms,
    round(stddev_exec_time::numeric, 2) AS stddev_ms
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 20;

-- Eng ko'p I/O qiladiganlar
SELECT 
    substring(query, 1, 80) AS query,
    calls,
    shared_blks_read,  -- disk read
    shared_blks_hit    -- cache hit
FROM pg_stat_statements
ORDER BY shared_blks_read DESC
LIMIT 20;

[databases]
myapp = host=postgres-primary.internal port=5432 dbname=myapp

[pgbouncer]
listen_port = 6432
listen_addr = 0.0.0.0

# Auth
auth_type = scram-sha-256
auth_file = /etc/pgbouncer/userlist.txt

# Pool config
pool_mode = transaction           # CRITICAL choice
max_client_conn = 1000            # clients PgBouncer'ga
default_pool_size = 20            # backends PG'ga (per db/user)
reserve_pool_size = 5             # emergency
reserve_pool_timeout = 3

# Timeouts
server_lifetime = 3600
server_idle_timeout = 600
query_timeout = 30
idle_transaction_timeout = 60

# Logging
log_connections = 0
log_disconnections = 0
stats_period = 60

from sqlalchemy.ext.asyncio import create_async_engine

# PgBouncer ORQALI connect qilamiz (6432 port)
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@pgbouncer.internal:6432/myapp",
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,           # connection tekshirish (health check)
    pool_recycle=3600,            # har soatda qayta yaratish
    connect_args={
        # MUHIM: transaction pool mode uchun
        "statement_cache_size": 0,
        "prepared_statement_cache_size": 0,
    }
)

-- PgBouncer admin'ga connect
-- psql -h pgbouncer -p 6432 pgbouncer

SHOW POOLS;
-- cl_active | cl_waiting | sv_active | sv_idle
--   45      |     0      |    18     |    2

SHOW STATS;
-- total_xact_count, total_query_count, avg_xact_time

SHOW CLIENTS;
-- clients va ularning state'i

# Primary postgresql.conf
wal_level = replica
max_wal_senders = 10
wal_keep_size = 2GB
synchronous_commit = on           # data loss tolerance

# Replica uchun user
CREATE ROLE replicator WITH REPLICATION LOGIN PASSWORD '...';

# Replica setup
pg_basebackup -h primary -D /var/lib/postgresql/data \
  -U replicator -P -W -R

# Replica automatic:
# standby.signal file yaratiladi
# primary_conninfo postgresql.auto.conf'ga yoziladi

"""Read-after-write pattern: just-updated data uchun primary'dan o'qish"""

class SmartDBRouter:
    async def update_and_track(self, user_id: str, data: dict):
        user = await self.primary.update_user(user_id, data)
        # Marker — keyingi 5 sek primary'dan o'qish
        await redis.setex(f"primary_only:{user_id}", 5, "1")
        return user
    
    async def read_user(self, user_id: str):
        # Check — just updated?
        if await redis.get(f"primary_only:{user_id}"):
            return await self.primary.get_user(user_id)
        
        # Lag check
        lag = await self.get_replica_lag()
        if lag > 10:  # 10 sek dan ko'p
            return await self.primary.get_user(user_id)
        
        return await self.replica.get_user(user_id)
    
    async def get_replica_lag(self) -> float:
        result = await self.replica.fetch_val(
            "SELECT EXTRACT(EPOCH FROM (NOW() - pg_last_xact_replay_timestamp()))"
        )
        return float(result or 0)

-- Citus extension (native PostgreSQL sharding)
CREATE EXTENSION citus;

-- Shardlash uchun jadval
CREATE TABLE events (
    id BIGSERIAL,
    tenant_id UUID NOT NULL,
    event_type TEXT,
    payload JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Shard key — tenant_id
SELECT create_distributed_table('events', 'tenant_id');

-- Citus avtomatik 32 shard yaratadi (default)
-- Query'lar avtomatik to'g'ri shard'ga boradi
-- Cross-shard aggregate: parallel scatter-gather

-- Reference table (kichik, har shard'da copy)
CREATE TABLE tenants (id UUID PRIMARY KEY, name TEXT);
SELECT create_reference_table('tenants');

-- Extension
CREATE EXTENSION vector;

-- Embedding jadval
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    content TEXT NOT NULL,
    embedding vector(1536),    -- OpenAI ada-002 dim
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- HNSW index (tez ANN search)
CREATE INDEX idx_documents_embedding 
  ON documents 
  USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);

-- ivfflat alternative (kichikroq RAM, lekin sekinroq)
-- CREATE INDEX ... USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);

-- Query — eng yaqin 10 ta document
SELECT 
    id, content, metadata,
    1 - (embedding <=> $1::vector) AS similarity
FROM documents
WHERE tenant_id = $2
ORDER BY embedding <=> $1::vector    -- cosine distance
LIMIT 10;

-- Operators:
-- <-> Euclidean distance
-- <=> Cosine distance  
-- <#> Inner product (negative)

from qdrant_client import QdrantClient, AsyncQdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue

client = AsyncQdrantClient(url="https://qdrant.internal:6333", api_key="...")

# Collection yaratish
await client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

# Index qo'shish (filter uchun)
await client.create_payload_index(
    collection_name="documents",
    field_name="tenant_id",
    field_schema="keyword"
)

# Batch insert
await client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=doc_id,
            vector=embedding,   # list[float] 1536
            payload={"tenant_id": str(tenant), "title": title, "content": text}
        )
        for doc_id, embedding, tenant, title, text in batch
    ]
)

# Hybrid query — vector + metadata filter
results = await client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[
            FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id)),
            FieldCondition(key="status", match=MatchValue(value="published"))
        ]
    ),
    limit=10,
    with_payload=True,
    score_threshold=0.7     # low-confidence'ni filter
)

# Qdrant da scalar quantization
from qdrant_client.models import ScalarQuantization, ScalarType, ScalarQuantizationConfig

await client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(
            type=ScalarType.INT8,
            quantile=0.99,
            always_ram=True,  # quantized vektorlar RAM'da
        )
    )
)
# RAM: 60 GB → 15 GB (4x saving)

from celery import Celery, Task
from celery.exceptions import MaxRetriesExceededError
import structlog

logger = structlog.get_logger()

app = Celery(
    'myapp',
    broker='redis://redis:6379/0',
    backend='redis://redis:6379/1'
)

app.conf.update(
    task_acks_late=True,              # faqat success'dan keyin ack
    task_reject_on_worker_lost=True,  # worker crash bo'lsa — qayta
    task_track_started=True,
    worker_prefetch_multiplier=1,     # fair distribution
    task_serializer='json',
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
    task_soft_time_limit=300,         # 5 daqiqa soft
    task_time_limit=360,              # 6 daqiqa hard kill
)

app.conf.task_routes = {
    'app.tasks.send_email':      {'queue': 'fast', 'priority': 5},
    'app.tasks.process_payment': {'queue': 'fast', 'priority': 9},
    'app.tasks.generate_report': {'queue': 'heavy', 'priority': 3},
    'app.tasks.embed_document':  {'queue': 'gpu', 'priority': 5},
}

class RetryableTask(Task):
    """Base class — automatic retry with exponential backoff"""
    autoretry_for = (ConnectionError, TimeoutError)
    retry_backoff = True
    retry_backoff_max = 300
    retry_jitter = True
    max_retries = 5

@app.task(
    base=RetryableTask,
    bind=True,
    rate_limit='100/m',
    acks_late=True,
)
def send_email(self, tenant_id: str, to: str, subject: str, body: str):
    logger.info("send_email.start", task_id=self.request.id, tenant=tenant_id)
    try:
        result = email_service.send(to=to, subject=subject, body=body)
        logger.info("send_email.success", task_id=self.request.id)
        return {"status": "sent", "message_id": result.id}
    except SMTPRateLimited as e:
        # 1 daqiqadan keyin qayta
        raise self.retry(exc=e, countdown=60)
    except SMTPRecipientsRefused as e:
        # Qayta urinib bo'lmaydi — DLQ ga
        logger.error("send_email.invalid_recipient", to=to)
        raise  # task_failure_handler DLQ'ga yozadi

from celery import chain, group, chord

# CHAIN — sequential (birin-ketin)
workflow = chain(
    fetch_user_data.s(user_id),
    enrich_with_analytics.s(),
    generate_report.s(),
    upload_to_s3.s(),
    notify_user.s(user_id)
)
workflow.apply_async()

# GROUP — parallel
header = group(
    process_image.s(img_id) for img_id in image_ids
)
result = header.apply_async()
results = result.join()  # wait all

# CHORD — parallel + callback
workflow = chord(
    [process_image.s(img) for img in images],  # parallel
    combine_images.s()                          # callback
)

# MAP-REDUCE style
reduce_workflow = chord(
    group(analyze_doc.s(doc) for doc in documents),
    aggregate_results.s()
)

from celery.schedules import crontab

app.conf.beat_schedule = {
    # Har 15 daqiqada
    'cleanup-expired-sessions': {
        'task': 'app.tasks.cleanup_sessions',
        'schedule': 900.0,
    },
    # Har kuni 02:00 UTC
    'daily-reports': {
        'task': 'app.tasks.generate_daily_reports',
        'schedule': crontab(hour=2, minute=0),
    },
    # Dushanba ertalab
    'weekly-digest': {
        'task': 'app.tasks.send_weekly_digest',
        'schedule': crontab(day_of_week=1, hour=9, minute=0),
    },
}

from aiokafka import AIOKafkaProducer
import json

class EventProducer:
    def __init__(self):
        self.producer = AIOKafkaProducer(
            bootstrap_servers='kafka1:9092,kafka2:9092,kafka3:9092',
            value_serializer=lambda v: json.dumps(v).encode(),
            key_serializer=lambda k: k.encode() if k else None,
            compression_type='lz4',
            acks='all',                    # all replicas confirm
            enable_idempotence=True,       # exactly-once
            max_in_flight_requests_per_connection=5,
        )
    
    async def start(self):
        await self.producer.start()
    
    async def emit(self, topic: str, event: dict, key: str = None):
        """Event'ni Kafka'ga yuborish"""
        # Envelope — standard metadata
        envelope = {
            "event_id": str(uuid4()),
            "event_type": event["type"],
            "event_version": "v1",
            "occurred_at": datetime.utcnow().isoformat(),
            "tenant_id": event.get("tenant_id"),
            "payload": event.get("payload", {}),
            "trace_id": get_current_trace_id(),
        }
        
        # key — same key → same partition → ordering
        await self.producer.send_and_wait(
            topic=topic,
            value=envelope,
            key=key or envelope.get("tenant_id"),
        )

# Ishlatish
producer = EventProducer()
await producer.start()

await producer.emit(
    topic="orders.events",
    event={
        "type": "order.created",
        "tenant_id": "tenant-123",
        "payload": {"order_id": "o-456", "amount": 99.99}
    },
    key="tenant-123"  # tenant bo'yicha ordering
)

from aiokafka import AIOKafkaConsumer

async def run_analytics_consumer():
    consumer = AIOKafkaConsumer(
        'orders.events',
        bootstrap_servers='kafka1:9092,kafka2:9092,kafka3:9092',
        group_id='analytics-v1',           # consumer group
        enable_auto_commit=False,           # manual commit muhim
        auto_offset_reset='earliest',
        value_deserializer=lambda v: json.loads(v.decode()),
        max_poll_records=100,
    )
    await consumer.start()
    
    try:
        async for msg in consumer:
            try:
                # Idempotency check — event'ni qayta ishlatmaslik
                if await already_processed(msg.value["event_id"]):
                    await consumer.commit()
                    continue
                
                # Process
                async with db.transaction():
                    await process_analytics_event(msg.value)
                    await mark_processed(msg.value["event_id"])
                
                # Manual commit — success'dan keyin
                await consumer.commit()
                
            except Exception as e:
                logger.error("consumer.error", error=str(e), offset=msg.offset)
                # Commit qilmaymiz — qayta keladi
                # N marta fail bo'lsa DLQ topic'ga yuboramiz
    finally:
        await consumer.stop()

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import List

@dataclass(frozen=True)
class WalletEvent:
    wallet_id: str
    event_type: str
    occurred_at: datetime
    version: int

@dataclass(frozen=True)
class WalletCreated(WalletEvent):
    user_id: str
    currency: str

@dataclass(frozen=True)
class MoneyDeposited(WalletEvent):
    amount: Decimal
    source: str

@dataclass(frozen=True)
class MoneyWithdrawn(WalletEvent):
    amount: Decimal
    destination: str

@dataclass
class Wallet:
    id: str
    user_id: str
    balance: Decimal = Decimal("0")
    version: int = 0
    
    @classmethod
    def from_events(cls, events: List[WalletEvent]) -> "Wallet":
        """State'ni event'lardan qayta tiklash"""
        if not events or not isinstance(events[0], WalletCreated):
            raise ValueError("Need WalletCreated first")
        
        first = events[0]
        wallet = cls(id=first.wallet_id, user_id=first.user_id, version=1)
        
        for event in events[1:]:
            wallet.apply(event)
        return wallet
    
    def apply(self, event: WalletEvent):
        if isinstance(event, MoneyDeposited):
            self.balance += event.amount
        elif isinstance(event, MoneyWithdrawn):
            if self.balance < event.amount:
                raise InsufficientFunds()
            self.balance -= event.amount
        self.version = event.version

# Event store
class EventStore:
    async def append(self, stream_id: str, events: List[WalletEvent], expected_version: int):
        """Optimistic concurrency control"""
        async with db.transaction():
            current = await db.fetch_val(
                "SELECT COALESCE(MAX(version), 0) FROM events WHERE stream_id = $1",
                stream_id
            )
            if current != expected_version:
                raise ConcurrencyError(f"Expected {expected_version}, got {current}")
            
            for event in events:
                await db.execute("""
                    INSERT INTO events (stream_id, event_type, payload, version, occurred_at)
                    VALUES ($1, $2, $3, $4, $5)
                """, stream_id, event.event_type, event_to_json(event), event.version, event.occurred_at)

"""Har service event publish qiladi, boshqalari subscribe qiladi"""

# Order service
async def create_order(user_id, items):
    order = await db.insert_order(user_id, items, status="pending")
    await events.publish("order.created", {
        "order_id": order.id, "user_id": user_id, 
        "items": items, "total": order.total
    })
    return order

# Payment service (subscribe to order.created)
async def handle_order_created(event):
    try:
        charge = await stripe.charge(event["user_id"], event["total"])
        await events.publish("payment.succeeded", {
            "order_id": event["order_id"], "charge_id": charge.id
        })
    except PaymentError:
        await events.publish("payment.failed", {
            "order_id": event["order_id"], "reason": str(e)
        })

# Inventory service (subscribe to payment.succeeded)
async def handle_payment_succeeded(event):
    try:
        await reserve_inventory(event["order_id"])
        await events.publish("inventory.reserved", {"order_id": event["order_id"]})
    except OutOfStock:
        await events.publish("inventory.failed", {"order_id": event["order_id"]})
        # Compensating action
        await events.publish("payment.refund_requested", {"order_id": event["order_id"]})

# Order service (subscribe to all)
async def handle_inventory_reserved(event):
    await db.update_order_status(event["order_id"], "confirmed")

async def handle_payment_failed(event):
    await db.update_order_status(event["order_id"], "cancelled")

"""Central coordinator step'larni boshqaradi"""

class OrderSaga:
    async def execute(self, order_data):
        saga_id = uuid4()
        state = {"order_id": None, "charge_id": None, "inventory_id": None}
        
        try:
            # Step 1: Create order
            order = await order_service.create(order_data)
            state["order_id"] = order.id
            
            # Step 2: Charge payment
            charge = await payment_service.charge(order.total)
            state["charge_id"] = charge.id
            
            # Step 3: Reserve inventory
            inventory = await inventory_service.reserve(order.items)
            state["inventory_id"] = inventory.id
            
            # Step 4: Confirm
            await order_service.confirm(order.id)
            return order
            
        except Exception as e:
            # Rollback — compensating transactions
            if state.get("inventory_id"):
                await inventory_service.release(state["inventory_id"])
            if state.get("charge_id"):
                await payment_service.refund(state["charge_id"])
            if state.get("order_id"):
                await order_service.cancel(state["order_id"])
            raise

-- Outbox jadvali — business DB'da
CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type TEXT NOT NULL,    -- 'order', 'user'
    aggregate_id UUID NOT NULL,
    event_type TEXT NOT NULL,
    payload JSONB NOT NULL,
    occurred_at TIMESTAMPTZ DEFAULT NOW(),
    published_at TIMESTAMPTZ,
    retry_count INT DEFAULT 0
);

CREATE INDEX idx_outbox_unpublished 
  ON outbox (occurred_at) 
  WHERE published_at IS NULL;

# Business logic — ONE transaction, DB + outbox
async def create_order(data):
    async with db.transaction():
        order = await db.insert_order(data)
        await db.insert_outbox({
            "aggregate_type": "order",
            "aggregate_id": order.id,
            "event_type": "order.created",
            "payload": order.to_dict(),
        })
    return order
    # ATOMIC: order va outbox birga commit bo'ladi

# Relay worker — outbox → Kafka
async def outbox_relay():
    while True:
        async with db.transaction():
            rows = await db.fetch_all("""
                SELECT * FROM outbox 
                WHERE published_at IS NULL 
                ORDER BY occurred_at
                LIMIT 100
                FOR UPDATE SKIP LOCKED
            """)
            
            for row in rows:
                try:
                    await kafka.send(
                        topic=f"{row.aggregate_type}.events",
                        value=row.payload,
                        key=str(row.aggregate_id),
                    )
                    await db.execute(
                        "UPDATE outbox SET published_at = NOW() WHERE id = $1",
                        row.id
                    )
                except Exception as e:
                    await db.execute(
                        "UPDATE outbox SET retry_count = retry_count + 1 WHERE id = $1",
                        row.id
                    )
        
        await asyncio.sleep(1)

# Alternative: Debezium (CDC) — PostgreSQL WAL'dan to'g'ridan to'g'ri o'qib Kafka'ga

from fastapi import WebSocket, WebSocketDisconnect
import redis.asyncio as redis

class ConnectionManager:
    """Multi-instance WebSocket via Redis pub-sub"""
    def __init__(self):
        self.local: dict[str, set[WebSocket]] = {}  # room_id -> connections
        self.redis = redis.Redis.from_url("redis://redis:6379")
        self.pubsub = self.redis.pubsub()
    
    async def connect(self, ws: WebSocket, room_id: str, user_id: str):
        await ws.accept()
        self.local.setdefault(room_id, set()).add(ws)
        
        # Redis subscription (agar bu birinchi connection shu room'ga)
        if len(self.local[room_id]) == 1:
            await self.pubsub.subscribe(f"room:{room_id}")
        
        # Join notification (boshqa instance'lar ham eshitadi)
        await self.redis.publish(f"room:{room_id}", json.dumps({
            "type": "user.joined", "user_id": user_id
        }))
    
    async def disconnect(self, ws: WebSocket, room_id: str):
        self.local.get(room_id, set()).discard(ws)
        if not self.local.get(room_id):
            await self.pubsub.unsubscribe(f"room:{room_id}")
    
    async def broadcast_to_room(self, room_id: str, message: dict):
        # Redis orqali — hamma instance'ga
        await self.redis.publish(f"room:{room_id}", json.dumps(message))
    
    async def redis_listener(self):
        """Redis'dan kelgan message'larni local WebSocket'larga forward"""
        async for message in self.pubsub.listen():
            if message["type"] != "message":
                continue
            room_id = message["channel"].decode().split(":")[1]
            data = json.loads(message["data"])
            
            for ws in self.local.get(room_id, set()):
                try:
                    await ws.send_json(data)
                except Exception:
                    pass  # connection broke

manager = ConnectionManager()

@app.on_event("startup")
async def startup():
    asyncio.create_task(manager.redis_listener())

@app.websocket("/ws/room/{room_id}")
async def ws_endpoint(ws: WebSocket, room_id: str):
    user = await authenticate_ws(ws)  # JWT in query param
    await manager.connect(ws, room_id, user.id)
    try:
        while True:
            data = await ws.receive_json()
            await manager.broadcast_to_room(room_id, {
                "type": "message",
                "user_id": user.id,
                "content": data["content"],
                "timestamp": time.time(),
            })
    except WebSocketDisconnect:
        await manager.disconnect(ws, room_id)

from fastapi import Request
from fastapi.responses import StreamingResponse
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

@app.post("/api/chat/stream")
async def chat_stream(request: Request, body: ChatRequest):
    async def event_stream():
        try:
            async with client.messages.stream(
                model="claude-opus-4-7",
                max_tokens=2048,
                system=body.system_prompt,
                messages=body.messages,
            ) as stream:
                async for text in stream.text_stream:
                    # Client disconnected?
                    if await request.is_disconnected():
                        break
                    
                    # SSE format: "data: \n\n"
                    yield f"data: {json.dumps({'type': 'token', 'text': text})}\n\n"
                
                # Final message
                final = await stream.get_final_message()
                yield f"data: {json.dumps({'type': 'done', 'usage': final.usage.dict()})}\n\n"
                
        except Exception as e:
            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
    
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",   # nginx buffering o'chirish
        }
    )

// Browser tomon — EventSource API
const eventSource = new EventSource('/api/chat/stream?q=hello');

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    if (data.type === 'token') {
        appendToUI(data.text);   // tokenlarni UI'da ko'rsatamiz
    } else if (data.type === 'done') {
        eventSource.close();
        console.log('Usage:', data.usage);
    }
};

eventSource.onerror = () => {
    // Browser avtomatik reconnect qiladi
    console.log('SSE error, retrying...');
};

import asyncio
import aiohttp
from asyncio import Semaphore, TaskGroup

# Pattern 1 — parallel requests with gather
async def fetch_all_fast(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [session.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in responses if not isinstance(r, Exception)]

# Pattern 2 — concurrency limit with Semaphore
async def fetch_with_limit(urls: list[str], limit: int = 10):
    sem = Semaphore(limit)
    
    async def bounded_fetch(url):
        async with sem:  # Max 10 parallel
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as r:
                    return await r.json()
    
    return await asyncio.gather(*[bounded_fetch(url) for url in urls])

# Pattern 3 — TaskGroup (Python 3.11+) — structured concurrency
async def process_user_data(user_id: str):
    async with TaskGroup() as tg:
        profile_task = tg.create_task(fetch_profile(user_id))
        orders_task = tg.create_task(fetch_orders(user_id))
        prefs_task = tg.create_task(fetch_preferences(user_id))
    # Bu yerga yetganda — hammasi yakunlangan
    # Birortasi xato bersa — TaskGroup hammasini cancel qiladi
    return {
        "profile": profile_task.result(),
        "orders": orders_task.result(),
        "preferences": prefs_task.result(),
    }

# Pattern 4 — async + process pool (CPU-bound)
from concurrent.futures import ProcessPoolExecutor

process_pool = ProcessPoolExecutor(max_workers=4)

async def process_image_async(image_path: str):
    loop = asyncio.get_event_loop()
    # CPU-bound ishni process pool'ga tashlaymiz
    result = await loop.run_in_executor(
        process_pool, 
        heavy_image_processing,
        image_path
    )
    return result

# Pattern 5 — timeout bilan
async def fetch_with_timeout(url: str):
    try:
        async with asyncio.timeout(5):  # Python 3.11+
            async with aiohttp.ClientSession() as s:
                async with s.get(url) as r:
                    return await r.json()
    except asyncio.TimeoutError:
        return None

# ✗ ANTI-PATTERN 1: blocking call async funcda
async def bad_1():
    response = requests.get("https://api.example.com")  # BLOCKS EVENT LOOP!
    return response.json()

# ✓ To'g'ri
async def good_1():
    async with aiohttp.ClientSession() as s:
        async with s.get("https://api.example.com") as r:
            return await r.json()


# ✗ ANTI-PATTERN 2: sync loop async'da
async def bad_2(users):
    results = []
    for user in users:
        r = await fetch_user(user)  # sequential!
        results.append(r)
    return results

# ✓ To'g'ri — parallel
async def good_2(users):
    return await asyncio.gather(*[fetch_user(u) for u in users])


# ✗ ANTI-PATTERN 3: CPU-bound async'da
async def bad_3(data):
    return heavy_computation(data)  # event loop bloklandi
    
# ✓ To'g'ri
async def good_3(data):
    return await loop.run_in_executor(process_pool, heavy_computation, data)


# ✗ ANTI-PATTERN 4: fire-and-forget without tracking
async def bad_4():
    asyncio.create_task(send_email(...))  # reference saqlanmadi — GC yeyishi mumkin!
    
# ✓ To'g'ri
bg_tasks = set()
async def good_4():
    task = asyncio.create_task(send_email(...))
    bg_tasks.add(task)
    task.add_done_callback(bg_tasks.discard)

from pydantic import BaseModel, Field, field_validator, model_validator, EmailStr
from pydantic import ConfigDict
from typing import Annotated, Literal
from decimal import Decimal
from datetime import datetime
from uuid import UUID

class OrderItem(BaseModel):
    model_config = ConfigDict(
        frozen=True,                   # immutable
        str_strip_whitespace=True,
        extra="forbid",                # noma'lum field → error
    )
    
    product_id: UUID
    quantity: Annotated[int, Field(gt=0, le=1000)]
    unit_price: Annotated[Decimal, Field(gt=0, decimal_places=2)]

class OrderCreate(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
    
    user_id: UUID
    items: Annotated[list[OrderItem], Field(min_length=1, max_length=100)]
    shipping_address: str
    currency: Literal["USD", "EUR", "GBP", "UZS"] = "USD"
    notes: str | None = Field(default=None, max_length=500)
    
    @field_validator("shipping_address")
    @classmethod
    def validate_address(cls, v: str) -> str:
        if len(v.split(",")) < 3:
            raise ValueError("Format: street, city, country")
        return v
    
    @model_validator(mode="after")
    def validate_total(self) -> "OrderCreate":
        total = sum(i.quantity * i.unit_price for i in self.items)
        if total > Decimal("100000"):
            raise ValueError("Order exceeds 100k — manual approval needed")
        return self

# Environment config — pydantic-settings
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_")
    
    database_url: str
    redis_url: str = "redis://localhost:6379"
    openai_api_key: str
    secret_key: str = Field(min_length=32)
    debug: bool = False
    allowed_origins: list[str] = []
    
    @field_validator("database_url")
    @classmethod
    def validate_db(cls, v: str) -> str:
        if not v.startswith(("postgresql://", "postgresql+asyncpg://")):
            raise ValueError("Must be PostgreSQL URL")
        return v

settings = Settings()  # avtomatik .env'dan o'qiydi

from typing import TypeVar, ParamSpec, Protocol, Self
from collections.abc import Awaitable, Callable

T = TypeVar("T")
P = ParamSpec("P")

# Generic class
class Repository[T]:
    async def find(self, id: UUID) -> T | None: ...
    async def save(self, entity: T) -> T: ...

# Protocol — structural typing (duck typing with types)
class Cacheable(Protocol):
    def cache_key(self) -> str: ...
    def ttl(self) -> int: ...

async def cache_item(item: Cacheable) -> None:
    key = item.cache_key()
    ttl = item.ttl()
    await redis.setex(key, ttl, item.model_dump_json())

# Self type (3.11+)
class QueryBuilder:
    def where(self, **kwargs) -> Self:
        self._filters.update(kwargs)
        return self  # typed as Self, so chaining works
    
    def order_by(self, field: str) -> Self:
        self._order = field
        return self

# Decorator with ParamSpec
def with_retry[T, **P](max_attempts: int = 3):
    def decorator(fn: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            for attempt in range(max_attempts):
                try:
                    return await fn(*args, **kwargs)
                except Exception:
                    if attempt == max_attempts - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)
        return wrapper
    return decorator

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import settings
from app.database import create_db_pool

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: resurslarni bir marta yaratamiz
    app.state.db_pool = await create_db_pool(settings.DATABASE_URL)
    app.state.redis = await create_redis_pool(settings.REDIS_URL)
    app.state.kafka = await create_kafka_producer(settings.KAFKA_URL)

    yield  # ← Shu yerda ilova ishlaydi

    # Shutdown: resurslarni toza yopamiz
    await app.state.kafka.stop()
    await app.state.redis.close()
    await app.state.db_pool.close()

from fastapi.middleware.cors import CORSMiddleware
from app.middleware import RequestIDMiddleware, LoggingMiddleware
from app.api.v1.router import api_router

app = FastAPI(
    title=settings.APP_NAME,
    version=settings.VERSION,
    lifespan=lifespan,
    docs_url="/docs" if settings.DEBUG else None,
)

# Tartib: oxirgi qo'shilgan — birinchi ishga tushadi (onion model)
app.add_middleware(RequestIDMiddleware)    # eng ichki
app.add_middleware(LoggingMiddleware)
app.add_middleware(CORSMiddleware,          # eng tashqi
    allow_origins=settings.ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
)

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

from fastapi import APIRouter, Depends, status
from app.models.schemas import OrderCreate, OrderResponse
from app.services.order_service import OrderService
from app.dependencies import get_order_service, get_current_user
from app.models.domain import User

router = APIRouter(prefix="/orders", tags=["orders"])

@router.post("", response_model=OrderResponse, status_code=201)
async def create_order(
    data: OrderCreate,
    current_user: User = Depends(get_current_user),
    service: OrderService = Depends(get_order_service),
) -> OrderResponse:
    order = await service.create_order(data=data, user=current_user)
    return OrderResponse.from_domain(order)

@router.get("", response_model=OrderListResponse)
async def list_orders(
    pagination: PaginationParams = Depends(get_pagination),
    current_user: User = Depends(get_current_user),
    service: OrderService = Depends(get_order_service),
) -> OrderListResponse:
    orders, total = await service.list_user_orders(
        user_id=current_user.id,
        limit=pagination.limit,
        offset=pagination.offset,
    )
    return OrderListResponse(
        data=[OrderResponse.from_domain(o) for o in orders],
        total=total,
        page=pagination.page,
    )

from app.repositories.order_repo import OrderRepository
from app.repositories.product_repo import ProductRepository
from app.repositories.wallet_repo import WalletRepository
from app.services.event_bus import EventBus
from app.models.domain import Order, User, OrderStatus
from app.models.schemas import OrderCreate
from app.exceptions import InsufficientStockError, InsufficientFundsError

class OrderService:
    def __init__(
        self,
        order_repo: OrderRepository,
        product_repo: ProductRepository,
        wallet_repo: WalletRepository,
        event_bus: EventBus,
    ):
        self.orders = order_repo
        self.products = product_repo
        self.wallets = wallet_repo
        self.events = event_bus

async def create_order(self, data: OrderCreate, user: User) -> Order:
    # 1. Mahsulot bor-yo'qligini tekshirish
    product = await self.products.get_by_id(data.product_id)
    if not product:
        raise NotFoundError(f"Product {data.product_id} not found")

    # 2. Omborda yetarli ekanligini tekshirish
    if product.stock < data.quantity:
        raise InsufficientStockError(
            f"Only {product.stock} available, requested {data.quantity}"
        )

    # 3. Narx va chegirma hisoblash
    total = product.price * data.quantity
    if user.is_premium:
        total *= Decimal("0.9")   # Premium foydalanuvchilarga 10% chegirma

    # 4. Hamyondagi pul yetarli ekanligini tekshirish
    wallet = await self.wallets.get_by_user(user.id)
    if wallet.balance < total:
        raise InsufficientFundsError(
            f"Balance: {wallet.balance}, required: {total}"
        )

    # 5. Transaction ichida saqlash (ATOMIC!)
    async with self.orders.transaction():
        order = await self.orders.create(
            user_id=user.id,
            product_id=data.product_id,
            quantity=data.quantity,
            total=total,
            status=OrderStatus.PENDING,
        )
        await self.products.decrement_stock(data.product_id, data.quantity)
        await self.wallets.debit(user.id, total)

    # 6. Async events — response foydalanuvchiga ketgandan keyin ishlaydi
    await self.events.publish("order.created", {
        "order_id": str(order.id),
        "user_id": str(user.id),
        "total": float(total),
    })

    return order

from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, func
from app.models.domain import Order, OrderStatus

class OrderRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    @asynccontextmanager
    async def transaction(self):
        """Atomic operation uchun transaction wrapper"""
        async with self.session.begin():
            yield

    async def create(
        self,
        user_id: UUID,
        product_id: UUID,
        quantity: int,
        total: Decimal,
        status: OrderStatus,
    ) -> Order:
        order = Order(
            user_id=user_id,
            product_id=product_id,
            quantity=quantity,
            total=total,
            status=status,
        )
        self.session.add(order)
        await self.session.flush()    # INSERT execute, lekin commit emas
        await self.session.refresh(order)   # DB'dan id, created_at ni olish
        return order

    async def list_by_user(
        self, user_id: UUID, limit: int = 20, offset: int = 0,
    ) -> tuple[list[Order], int]:
        # Data query
        data_query = (
            select(Order)
            .where(Order.user_id == user_id)
            .order_by(Order.created_at.desc())
            .limit(limit).offset(offset)
        )
        # Count query (pagination uchun total kerak)
        count_query = (
            select(func.count(Order.id))
            .where(Order.user_id == user_id)
        )

        # PARALLEL execute — ikkala query bir vaqtda
        results = await asyncio.gather(
            self.session.execute(data_query),
            self.session.execute(count_query),
        )

        orders = list(results[0].scalars().all())
        total = results[1].scalar_one()
        return orders, total

from fastapi import Depends, Request
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import SessionLocal

async def get_db() -> AsyncSession:
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import HTTPException
import jwt
from app.config import settings

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db),
) -> User:
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.JWT_SECRET,
            algorithms=[settings.JWT_ALGORITHM],
        )
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "Invalid token")

    user = await db.get(User, UUID(payload["sub"]))
    if not user:
        raise HTTPException(401, "User not found")
    return user

async def get_order_repo(db: AsyncSession = Depends(get_db)) -> OrderRepository:
    return OrderRepository(db)

async def get_product_repo(db: AsyncSession = Depends(get_db)) -> ProductRepository:
    return ProductRepository(db)

async def get_wallet_repo(db: AsyncSession = Depends(get_db)) -> WalletRepository:
    return WalletRepository(db)

async def get_order_service(
    order_repo: OrderRepository = Depends(get_order_repo),
    product_repo: ProductRepository = Depends(get_product_repo),
    wallet_repo: WalletRepository = Depends(get_wallet_repo),
    event_bus: EventBus = Depends(lambda r: r.app.state.event_bus),
) -> OrderService:
    return OrderService(
        order_repo=order_repo,
        product_repo=product_repo,
        wallet_repo=wallet_repo,
        event_bus=event_bus,
    )

from pydantic import BaseModel, Field
from uuid import UUID

class OrderCreate(BaseModel):
    """Client'dan kelayotgan data — faqat kerakli fieldlar"""
    product_id: UUID
    quantity: int = Field(..., ge=1, le=100, description="1-100 oralig'ida")

from pydantic import ConfigDict
from datetime import datetime
from decimal import Decimal
from typing import Self
from app.models.domain import Order, OrderStatus

class OrderResponse(BaseModel):
    """Client'ga qaytariladigan data"""
    model_config = ConfigDict(from_attributes=True)

    id: UUID
    product_id: UUID
    quantity: int
    total: Decimal
    status: OrderStatus
    created_at: datetime
    # user_id qaytarilmaydi — user o'ziniki ekanini biladi
    # internal_notes qaytarilmaydi — bu admin uchun

    @classmethod
    def from_domain(cls, order: Order) -> Self:
        return cls.model_validate(order)

import pytest
from decimal import Decimal
from unittest.mock import AsyncMock
from app.services.order_service import OrderService
from app.models.schemas import OrderCreate
from app.models.domain import Product, User, Wallet
from app.exceptions import InsufficientStockError

@pytest.fixture
def mock_dependencies():
    """Barcha dependencies — fake obyektlar"""
    return {
        "order_repo": AsyncMock(),
        "product_repo": AsyncMock(),
        "wallet_repo": AsyncMock(),
        "event_bus": AsyncMock(),
    }

@pytest.fixture
def service(mock_dependencies):
    return OrderService(**mock_dependencies)

async def test_success_with_enough_stock_and_balance(
    service, mock_dependencies
):
    # Arrange — fake ma'lumotlarni tayyorlash
    product = Product(id="p1", price=Decimal("100"), stock=10)
    wallet = Wallet(user_id="u1", balance=Decimal("500"))
    user = User(id="u1", is_premium=False)

    mock_dependencies["product_repo"].get_by_id.return_value = product
    mock_dependencies["wallet_repo"].get_by_user.return_value = wallet

    # Act — tekshiriladigan amalni bajarish
    data = OrderCreate(product_id="p1", quantity=3)
    order = await service.create_order(data, user)

    # Assert — kutilgan natijani tekshirish
    assert order.quantity == 3
    mock_dependencies["product_repo"].decrement_stock.assert_called_with("p1", 3)
    mock_dependencies["wallet_repo"].debit.assert_called_with("u1", Decimal("300"))
    mock_dependencies["event_bus"].publish.assert_called_once()

async def test_raises_if_stock_insufficient(service, mock_dependencies):
    mock_dependencies["product_repo"].get_by_id.return_value = Product(
        id="p1", price=Decimal("100"), stock=2   # faqat 2 dona bor
    )

    # Quantity 5 — stock'dan ko'p
    with pytest.raises(InsufficientStockError):
        await service.create_order(
            OrderCreate(product_id="p1", quantity=5),
            User(id="u1", is_premium=False),
        )

async def test_premium_user_gets_10_percent_discount(
    service, mock_dependencies
):
    mock_dependencies["product_repo"].get_by_id.return_value = Product(
        id="p1", price=Decimal("100"), stock=10
    )
    mock_dependencies["wallet_repo"].get_by_user.return_value = Wallet(
        user_id="u1", balance=Decimal("1000")
    )
    user = User(id="u1", is_premium=True)   # ← PREMIUM user

    await service.create_order(OrderCreate(product_id="p1", quantity=2), user)

    # 100 * 2 * 0.9 = 180 (not 200)
    mock_dependencies["wallet_repo"].debit.assert_called_with(
        "u1", Decimal("180.0")
    )

# syntax=docker/dockerfile:1.7

# Stage 1: builder
FROM python:3.12-slim AS builder

WORKDIR /build

# System deps faqat build uchun
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# uv — yangi tez package manager (2026 default)
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uv

# Dependencies — alohida layer (cache optimization)
COPY pyproject.toml uv.lock ./
RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --frozen --no-dev --no-install-project

# Stage 2: runtime
FROM python:3.12-slim AS runtime

# Non-root user
RUN groupadd -r app && useradd -r -g app -d /app -s /bin/bash app

WORKDIR /app

# System deps (minimal)
RUN apt-get update && apt-get install -y --no-install-recommends \
    libpq5 curl \
    && rm -rf /var/lib/apt/lists/* \
    && apt-get clean

# Dependencies builder'dan ko'chiriladi
COPY --from=builder /build/.venv /app/.venv
ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

# App code
COPY --chown=app:app . .

USER app
EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: api
  labels: { app: api, tier: backend }
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0       # downtime yo'q
  selector:
    matchLabels: { app: api }
  template:
    metadata:
      labels: { app: api, tier: backend }
    spec:
      topologySpreadConstraints:
        - maxSkew: 1
          topologyKey: kubernetes.io/hostname
          whenUnsatisfiable: ScheduleAnyway
          labelSelector:
            matchLabels: { app: api }
      containers:
      - name: api
        image: registry.internal/api:v1.42.0
        ports: [{ containerPort: 8000 }]
        
        # Resource limits — MUST have
        resources:
          requests: { cpu: "250m", memory: "512Mi" }
          limits:   { cpu: "1000m", memory: "1Gi" }
        
        # Probes — 3 ta turi
        startupProbe:
          httpGet: { path: /health, port: 8000 }
          failureThreshold: 30
          periodSeconds: 2
        livenessProbe:
          httpGet: { path: /health, port: 8000 }
          periodSeconds: 15
          failureThreshold: 3
        readinessProbe:
          httpGet: { path: /ready, port: 8000 }
          periodSeconds: 5
          failureThreshold: 2
        
        env:
          - name: DATABASE_URL
            valueFrom: { secretKeyRef: { name: db, key: url } }
          - name: POD_NAME
            valueFrom: { fieldRef: { fieldPath: metadata.name } }
        
        lifecycle:
          preStop:
            exec:
              command: ["sh", "-c", "sleep 15"]   # graceful drain

---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata: { name: api }
spec:
  scaleTargetRef: { apiVersion: apps/v1, kind: Deployment, name: api }
  minReplicas: 3
  maxReplicas: 20
  metrics:
    - type: Resource
      resource: { name: cpu, target: { type: Utilization, averageUtilization: 70 } }
    - type: Pods
      pods:
        metric: { name: http_requests_per_second }
        target: { type: AverageValue, averageValue: "100" }
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300   # 5 daqiqa kutib tursin

name: deploy
on:
  push: { branches: [main] }

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16-alpine
        env: { POSTGRES_PASSWORD: test }
        options: --health-cmd "pg_isready"
      redis:
        image: redis:7-alpine
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v3
      - run: uv sync --frozen
      - run: uv run ruff check .
      - run: uv run mypy app/
      - run: uv run pytest --cov=app --cov-fail-under=80
  
  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: aquasecurity/trivy-action@master
        with: { scan-type: 'fs', exit-code: '1', severity: 'CRITICAL,HIGH' }
      - uses: snyk/actions/python@master
        env: { SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }} }
  
  build:
    needs: [test, security]
    runs-on: ubuntu-latest
    permissions: { contents: read, packages: write, id-token: write }
    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/metadata-action@v5
        id: meta
        with:
          images: ghcr.io/${{ github.repository }}
          tags: |
            type=sha,format=long
            type=ref,event=branch
      - uses: docker/build-push-action@v5
        with:
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          cache-from: type=gha
          cache-to: type=gha,mode=max
          provenance: true       # SLSA provenance
          sbom: true             # Software Bill of Materials
  
  deploy:
    needs: build
    runs-on: ubuntu-latest
    environment: production
    steps:
      - uses: actions/checkout@v4
      - uses: azure/setup-kubectl@v4
      - run: |
          kubectl set image deployment/api \
            api=ghcr.io/${{ github.repository }}:sha-${{ github.sha }}
          kubectl rollout status deployment/api --timeout=5m

import structlog
import logging

structlog.configure(
    processors=[
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.dict_tracebacks,
        structlog.processors.JSONRenderer(),  # JSON output
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

# Ishlatish — ALWAYS with context
logger.info(
    "order.created",
    order_id=str(order.id),
    tenant_id=str(order.tenant_id),
    user_id=str(order.user_id),
    amount=float(order.total),
    currency=order.currency,
    trace_id=get_current_trace_id(),
)

# Output (JSON):
# {"event": "order.created", "order_id": "...", "tenant_id": "...",
#  "timestamp": "2026-04-21T10:23:45Z", "level": "info", ...}

from prometheus_client import Counter, Histogram, Gauge, Info

# RED method: Rate, Errors, Duration
http_requests = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status', 'tenant']
)

http_duration = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint'],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)

# USE method: Utilization, Saturation, Errors
active_connections = Gauge(
    'db_connections_active',
    'Active DB connections',
    ['pool']
)

queue_depth = Gauge('celery_queue_depth', 'Tasks in queue', ['queue'])

# Business metrics
orders_created = Counter(
    'orders_created_total',
    'Orders created',
    ['tenant', 'plan', 'currency']
)

app_info = Info('app_version', 'Application version')
app_info.info({'version': '1.42.0', 'build': 'a1b2c3'})

from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Setup
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://otel-collector:4317")
))
trace.set_tracer_provider(provider)

# Auto-instrument frameworks
FastAPIInstrumentor.instrument_app(app)
SQLAlchemyInstrumentor().instrument(engine=engine)
HTTPXClientInstrumentor().instrument()

# Custom span'lar
tracer = trace.get_tracer(__name__)

@app.post("/orders")
async def create_order(data: OrderCreate):
    with tracer.start_as_current_span("create_order") as span:
        span.set_attribute("user.id", str(data.user_id))
        span.set_attribute("order.items_count", len(data.items))
        
        with tracer.start_as_current_span("validate_inventory"):
            await check_inventory(data.items)
        
        with tracer.start_as_current_span("charge_payment"):
            charge = await stripe_charge(data.total)
            span.set_attribute("charge.id", charge.id)
        
        with tracer.start_as_current_span("save_order"):
            order = await db.save_order(data)
        
        span.set_attribute("order.id", str(order.id))
        return order

from anthropic import AsyncAnthropic

client = AsyncAnthropic()

response = await client.messages.create(
    model="claude-opus-4-7",          # model tanlovi
    max_tokens=1024,                   # response limit (to'xtash uchun)
    temperature=0.7,                   # 0.0 = deterministic, 1.0 = creative
    top_p=0.95,                        # nucleus sampling
    system="Sen professional Python dasturchisisan.",  # persona
    messages=[
        {"role": "user", "content": "Fibonacci function yoz."}
    ],
)

# Key parametrlar:
# temperature — kreativlik darajasi
#   0.0: har safar bir xil javob (code, facts)
#   0.3-0.7: professional ishlar
#   1.0+: kreativ yozish, brainstorm
# 
# top_p — probability mass
#   0.9: faqat top 90% ehtimolli token'lar
# 
# max_tokens — qancha token generate qilsin
#   Kengroq context → qimmat va sekinroq

# PATTERN 1 — Role + Task + Format + Examples
SYSTEM_PROMPT = """Sen senior Python backend dasturchi va code reviewer.

Vazifang:
- Berilgan Python kodni ko'rib chiqish
- Security, performance, best practices muammolarini topish  
- Har muammo uchun tuzatilgan versiya berish

Javob formati (JSON):
{
  "issues": [
    {
      "severity": "critical|high|medium|low",
      "category": "security|performance|style|bug",
      "description": "...",
      "line": 15,
      "suggested_fix": "..."
    }
  ],
  "summary": "umumiy baholash"
}

Misollar:
- SQL injection topsang → severity: "critical", category: "security"
- N+1 query topsang → severity: "high", category: "performance"
"""

# PATTERN 2 — Structured output with Pydantic
from pydantic import BaseModel

class Issue(BaseModel):
    severity: Literal["critical", "high", "medium", "low"]
    category: Literal["security", "performance", "style", "bug"]
    description: str
    line: int
    suggested_fix: str

class ReviewResult(BaseModel):
    issues: list[Issue]
    summary: str

# Instructor library bilan (Pydantic validate)
import instructor
client = instructor.from_anthropic(AsyncAnthropic())

result: ReviewResult = await client.messages.create(
    model="claude-opus-4-7",
    response_model=ReviewResult,  # auto-validate
    messages=[{"role": "user", "content": code_to_review}]
)

# PATTERN 3 — Chain of Thought
COT_PROMPT = """Muammo: {problem}

Qadam-qadam o'ylang:
1. Masala nima haqida?
2. Qanday ma'lumotlar berilgan?
3. Qanday yondashuvlar mavjud?
4. Har birini qisqacha baholang
5. Eng yaxshi yondashuvni tanlang va bajaring

Xulosa:"""

"""Model tool'larni chaqirib real ishlarni bajara oladi"""

tools = [
    {
        "name": "get_weather",
        "description": "Shaharning ob-havo holatini olish",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "Shahar nomi"},
                "units": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["city"]
        }
    },
    {
        "name": "search_database",
        "description": "Ichki ma'lumotlar bazasida qidirish",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "limit": {"type": "integer", "default": 10}
            },
            "required": ["query"]
        }
    }
]

async def chat_with_tools(user_message: str):
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        response = await client.messages.create(
            model="claude-opus-4-7",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )
        
        if response.stop_reason == "end_turn":
            return response.content[0].text
        
        if response.stop_reason == "tool_use":
            # Model tool chaqirishni so'radi
            messages.append({"role": "assistant", "content": response.content})
            
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = await execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })
            
            messages.append({"role": "user", "content": tool_results})
            # Loop — model tool natijasi bilan keyingi qadam

from openai import AsyncOpenAI
from qdrant_client import AsyncQdrantClient

openai = AsyncOpenAI()
qdrant = AsyncQdrantClient(url="...")

# ─── INDEXING (offline, once) ───
async def index_document(doc_id: str, content: str, metadata: dict):
    # 1. Chunking
    chunks = chunk_text(content, chunk_size=500, overlap=50)
    
    # 2. Embedding (batch)
    embeddings = await openai.embeddings.create(
        model="text-embedding-3-small",  # 1536 dim
        input=chunks
    )
    
    # 3. Store
    points = [
        {"id": f"{doc_id}_{i}", 
         "vector": emb.embedding,
         "payload": {"doc_id": doc_id, "chunk_index": i, "text": chunk, **metadata}}
        for i, (chunk, emb) in enumerate(zip(chunks, embeddings.data))
    ]
    await qdrant.upsert(collection_name="kb", points=points)

# ─── QUERY (real-time) ───
async def answer_question(question: str, tenant_id: str) -> dict:
    # 1. Embed question (same model as indexing!)
    query_emb = (await openai.embeddings.create(
        model="text-embedding-3-small",
        input=[question]
    )).data[0].embedding
    
    # 2. Retrieve top-k
    results = await qdrant.search(
        collection_name="kb",
        query_vector=query_emb,
        query_filter={"must": [{"key": "tenant_id", "match": {"value": tenant_id}}]},
        limit=5,
        score_threshold=0.7  # past score'larni filter
    )
    
    # 3. Build context
    context = "\n\n".join([
        f"[Source {i+1}] {r.payload['text']}"
        for i, r in enumerate(results)
    ])
    
    # 4. LLM generation
    response = await openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": 
             "Javob bering FAQAT berilgan context asosida. "
             "Agar context yetarli emas — 'Ma'lumot yo'q' deb yozing. "
             "Har faktdan keyin [Source N] manba ko'rsating."},
            {"role": "user", "content": f"Context:\n{context}\n\nSavol: {question}"}
        ],
        temperature=0.1  # faktual javob uchun past
    )
    
    return {
        "answer": response.choices[0].message.content,
        "sources": [{"text": r.payload["text"], "score": r.score, "doc_id": r.payload["doc_id"]} 
                    for r in results]
    }

"""Hybrid = BM25 (keyword) + Vector (semantic) + RRF fusion"""

async def hybrid_search(question: str, k: int = 20) -> list[dict]:
    # 1. Vector search — semantic
    query_emb = await embed(question)
    vector_results = await qdrant.search(
        collection_name="kb",
        query_vector=query_emb,
        limit=k
    )
    
    # 2. BM25 search — keyword (PostgreSQL yoki Elasticsearch)
    bm25_results = await db.fetch_all("""
        SELECT id, chunk, doc_id,
               ts_rank_cd(search_vector, query) AS score
        FROM knowledge_chunks,
             plainto_tsquery('english', $1) query
        WHERE search_vector @@ query
        ORDER BY score DESC
        LIMIT $2
    """, question, k)
    
    # 3. Reciprocal Rank Fusion (RRF) — 2 ro'yxatni birlashtirish
    scores = {}
    RRF_K = 60
    
    for rank, r in enumerate(vector_results):
        scores[r.id] = scores.get(r.id, 0) + 1.0 / (RRF_K + rank)
    
    for rank, r in enumerate(bm25_results):
        scores[r["id"]] = scores.get(r["id"], 0) + 1.0 / (RRF_K + rank)
    
    # 4. Sort va return
    top_ids = sorted(scores.keys(), key=lambda i: scores[i], reverse=True)[:10]
    return [await get_chunk(id) for id in top_ids]


async def rag_with_rerank(question: str) -> dict:
    # 1. Hybrid retrieval — over-fetch (20)
    candidates = await hybrid_search(question, k=20)
    
    # 2. Reranker — cross-encoder (expensive lekin aniq)
    #    Query + chunk'ni bitta model'ga yuboramiz, aniqroq score
    from cohere import AsyncClient as CohereClient
    cohere = CohereClient(...)
    
    rerank_result = await cohere.rerank(
        query=question,
        documents=[c["text"] for c in candidates],
        model="rerank-english-v3.0",
        top_n=5
    )
    
    # 3. Top 5 — eng sifatli context
    reranked = [candidates[r.index] for r in rerank_result.results]
    
    # 4. LLM
    return await generate_with_context(question, reranked)

CONTEXTUALIZE_PROMPT = """Quyida to'liq hujjat va undan olingan qism.
Berilgan qism hujjatning qaysi qismiga tegishli, qanday context'da ekanini qisqa 
50-100 so'zda tushuntiring.

Full document:
{full_doc}

Chunk:
{chunk}

Context (faqat qisqa izoh, hech narsa qo'shmang):"""

async def contextualize_chunk(full_doc: str, chunk: str) -> str:
    """Har chunk'dan oldin qo'shiladigan kontekst izoh"""
    response = await client.messages.create(
        model="claude-haiku-4-5-20251001",  # kichik model — arzon
        max_tokens=150,
        messages=[{
            "role": "user", 
            "content": CONTEXTUALIZE_PROMPT.format(full_doc=full_doc[:50000], chunk=chunk)
        }]
    )
    return response.content[0].text

async def index_with_context(doc_id: str, content: str):
    chunks = chunk_text(content, 500)
    
    # Har chunk'ni kontekstualizatsiya qilamiz
    contextualized = await asyncio.gather(*[
        contextualize_chunk(content, c) for c in chunks
    ])
    
    # Chunk = context + original
    final_chunks = [f"{ctx}\n\n{chunk}" for ctx, chunk in zip(contextualized, chunks)]
    
    # Embed va store
    embeddings = await embed_batch(final_chunks)
    await qdrant.upsert(...)

"""Agent retrieval'ni tool sifatida ishlatadi"""

TOOLS = [
    {
        "name": "search_knowledge",
        "description": "Ichki KB'dan qidirish. Multiple tadqiqot savollari uchun bir necha marta chaqirish mumkin.",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}, "limit": {"type": "integer", "default": 5}},
            "required": ["query"]
        }
    },
    {
        "name": "search_web",
        "description": "Yangi tashqi ma'lumot kerak bo'lsa web search",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    }
]

AGENTIC_SYSTEM = """Sen research agentsan. Murakkab savollarga javob berish uchun:
1. Savolni sub-savollarga bo'l
2. Har biri uchun search_knowledge yoki search_web ishlat
3. Yetarli ma'lumot yig'ilgach — sintez qil
4. MAX 5 ta search iteratsiyasi"""

async def agentic_rag_answer(question: str, max_iterations: int = 5):
    messages = [{"role": "user", "content": question}]
    iteration = 0
    
    while iteration < max_iterations:
        response = await client.messages.create(
            model="claude-opus-4-7",
            max_tokens=2000,
            tools=TOOLS,
            system=AGENTIC_SYSTEM,
            messages=messages
        )
        
        if response.stop_reason == "end_turn":
            return response.content[0].text
        
        messages.append({"role": "assistant", "content": response.content})
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                if block.name == "search_knowledge":
                    results = await hybrid_search(block.input["query"])
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(results[:5])
                    })
        
        messages.append({"role": "user", "content": tool_results})
        iteration += 1
    
    # Budget tugadi — hozirgi ma'lumot bilan javob
    messages.append({"role": "user", "content": "Budget tugadi. Hozirgi ma'lumot asosida javob ber."})
    final = await client.messages.create(model="claude-opus-4-7", max_tokens=2000, messages=messages)
    return final.content[0].text

from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevance
from datasets import Dataset

# Ground truth dataset
eval_data = Dataset.from_dict({
    "question": [...],           # savollar
    "answer": [...],             # sizning RAG javoblari
    "contexts": [...],           # retrieved chunks
    "ground_truth": [...]        # to'g'ri javoblar (annotator)
})

result = evaluate(
    eval_data,
    metrics=[context_precision, context_recall, faithfulness, answer_relevance]
)
print(result)
# {'context_precision': 0.82, 'context_recall': 0.78, ...}

# OpenAI — dimensions parameter
response = await openai.embeddings.create(
    model="text-embedding-3-large",
    input=["hello"],
    dimensions=512    # 3072 → 512 ga truncate
)

# Yoki oddiy indexing:
# Coarse search: 256-dim (tez)
# Rerank: to'liq 3072-dim (aniq)

# Option 1 — Cohere Rerank (managed)
from cohere import AsyncClient

cohere = AsyncClient(api_key="...")

async def cohere_rerank(query: str, docs: list[str], top_n: int = 5):
    result = await cohere.rerank(
        query=query,
        documents=docs,
        model="rerank-multilingual-v3.0",  # 100+ tillar
        top_n=top_n
    )
    return [(r.index, r.relevance_score) for r in result.results]

# Option 2 — BGE-reranker (self-host)
from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-v2-m3', use_fp16=True)

def local_rerank(query: str, docs: list[str], top_n: int = 5):
    pairs = [[query, doc] for doc in docs]
    scores = reranker.compute_score(pairs, normalize=True)
    ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)[:top_n]
    return ranked

# Option 3 — Jina reranker
from jina_ai_search_foundation_sdk import JinaRerank

"""QLoRA fine-tuning Llama 3.1 8B on custom domain data"""
from unsloth import FastLanguageModel
from trl import SFTTrainer
from transformers import TrainingArguments
from datasets import load_dataset
import torch

# 1. Model yuklash (4-bit quantized)
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="unsloth/Meta-Llama-3.1-8B-Instruct",
    max_seq_length=2048,
    dtype=None,
    load_in_4bit=True,  # QLoRA — NF4 quantization
)

# 2. LoRA adapter qo'shish
model = FastLanguageModel.get_peft_model(
    model,
    r=16,              # rank — capacity vs overfitting tradeoff
    target_modules=[
        "q_proj", "k_proj", "v_proj", "o_proj",  # attention
        "gate_proj", "up_proj", "down_proj"       # MLP
    ],
    lora_alpha=16,     # scaling — odatda r ga teng
    lora_dropout=0,
    bias="none",
    use_gradient_checkpointing="unsloth",  # RAM tejash
    random_state=42,
)

# 3. Dataset prep — instruction format
def format_example(example):
    return {
        "text": f"""### Instruction:
{example['instruction']}

### Input:
{example.get('input', '')}

### Response:
{example['output']}"""
    }

dataset = load_dataset("json", data_files="training_data.jsonl", split="train")
dataset = dataset.map(format_example)

# 4. Training args
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    dataset_text_field="text",
    max_seq_length=2048,
    args=TrainingArguments(
        output_dir="outputs",
        per_device_train_batch_size=2,
        gradient_accumulation_steps=4,   # effective batch = 8
        num_train_epochs=3,
        learning_rate=2e-4,              # LoRA — 1e-4 to 3e-4
        warmup_ratio=0.03,
        lr_scheduler_type="cosine",
        logging_steps=10,
        save_steps=100,
        fp16=not torch.cuda.is_bf16_supported(),
        bf16=torch.cuda.is_bf16_supported(),
        optim="adamw_8bit",              # memory-efficient
        seed=42,
    ),
)

# 5. Train!
trainer.train()

# 6. Save adapter (kichik — ~200 MB)
model.save_pretrained("my-finetuned-adapter")
tokenizer.save_pretrained("my-finetuned-adapter")

# 7. Merge va GGUF export (Ollama uchun)
model.save_pretrained_merged("merged_model", tokenizer, save_method="merged_16bit")
model.save_pretrained_gguf("gguf_model", tokenizer, quantization_method="q4_k_m")

{"instruction": "Berilgan Python kodda xato topib, tuzatilgan versiyani bering.",
 "input": "def divide(a, b): return a / b",
 "output": "```python\ndef divide(a: float, b: float) -> float:\n    if b == 0:\n        raise ValueError('Cannot divide by zero')\n    return a / b\n```\n\nTuzatish: `ZeroDivisionError` himoyasi va type hints qo'shildi."}
{"instruction": "SQL query'ni optimallashtiring.",
 "input": "SELECT * FROM orders WHERE UPPER(email) = 'TEST@EXAMPLE.COM'",
 "output": "..."}

# Ollama install
curl -fsSL https://ollama.com/install.sh | sh

# Modelfile yaratish (fine-tuned modelni Ollama'ga qo'shish)
cat > Modelfile << 'EOF'
FROM ./gguf_model/unsloth.Q4_K_M.gguf

TEMPLATE """### Instruction:
{{ .Prompt }}

### Response:
"""

PARAMETER temperature 0.7
PARAMETER top_p 0.9
PARAMETER num_ctx 2048
EOF

# Model create
ollama create my-assistant -f Modelfile

# Ishga tushirish
ollama run my-assistant

# API sifatida (OpenAI-compatible)
# http://localhost:11434/v1/chat/completions

from openai import OpenAI

# Ollama OpenAI-compatible API'ni qo'llab-quvvatlaydi
client = OpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama"  # ignored
)

response = client.chat.completions.create(
    model="my-assistant",
    messages=[
        {"role": "user", "content": "SQL query'ni qanday optimallashtirish?"}
    ]
)
print(response.choices[0].message.content)

"""ReAct: Thought → Action → Observation loop"""

REACT_PROMPT = """Sen agentsan. Berilgan vazifani bajarish uchun tool'lardan foydalanasan.

Har qadamda:
Thought: nima qilmoqchiligingni o'yla
Action: tool chaqirish
Observation: tool natijasi (avtomatik to'ldiriladi)

Yakuniy javob bo'lganda:
Final Answer: to'liq javob

Tools:
- search_docs(query): ichki hujjatlarda qidirish
- fetch_url(url): sahifa matnini olish
- calculate(expression): hisoblash
- query_db(sql): ma'lumotlar bazasidan

Vazifa: {task}"""

async def react_agent(task: str, max_steps: int = 10):
    messages = [{"role": "user", "content": REACT_PROMPT.format(task=task)}]
    
    for step in range(max_steps):
        response = await client.messages.create(
            model="claude-opus-4-7",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
        )
        
        messages.append({"role": "assistant", "content": response.content})
        
        # Terminal condition
        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text" and "Final Answer:" in block.text:
                    return block.text.split("Final Answer:")[-1].strip()
        
        # Tool execution
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    try:
                        result = await execute_tool(block.name, block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result)[:5000]  # truncate
                        })
                    except Exception as e:
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": f"Error: {str(e)}",
                            "is_error": True
                        })
            messages.append({"role": "user", "content": tool_results})
    
    return "Max steps reached without final answer"

"""Supervisor agent sub-agent'larga vazifalar tarqatadi"""
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal

class AgentState(TypedDict):
    task: str
    research: str
    analysis: str
    final_report: str
    next_step: str

# Specialist agents
async def researcher_agent(state: AgentState):
    """Ma'lumot yig'ish"""
    result = await run_agent(
        system="Sen researcher. Faqat faktlarni yig'. Source cite qil.",
        tools=[search_web, search_docs],
        task=state["task"]
    )
    return {"research": result}

async def analyst_agent(state: AgentState):
    """Tahlil va insight"""
    result = await run_agent(
        system="Sen data analyst. Raqamlarni tushuntir, trend'larni top.",
        tools=[calculate, query_db],
        task=f"Research:\n{state['research']}\n\nTahlil qil."
    )
    return {"analysis": result}

async def writer_agent(state: AgentState):
    """Final report yozish"""
    result = await run_agent(
        system="Sen technical writer. Aniq, qisqa, strukturali yoz.",
        task=f"Research:\n{state['research']}\n\nAnalysis:\n{state['analysis']}\n\nReport yoz."
    )
    return {"final_report": result}

# Supervisor
async def supervisor(state: AgentState) -> dict:
    """Supervisor keyingi kimni chaqirishni aniqlaydi"""
    response = await client.messages.create(
        model="claude-opus-4-7",
        system="Sen supervisor. Berilgan state'ga qarab keyingi qadamni tanla: "
               "researcher, analyst, writer, yoki done.",
        messages=[{"role": "user", "content": json.dumps(state)}]
    )
    next_step = response.content[0].text.strip().lower()
    return {"next_step": next_step}

def route(state: AgentState) -> Literal["researcher", "analyst", "writer", END]:
    return state["next_step"] if state["next_step"] != "done" else END

# Graph tuzish
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher_agent)
graph.add_node("analyst", analyst_agent)
graph.add_node("writer", writer_agent)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route)
graph.add_edge("researcher", "supervisor")
graph.add_edge("analyst", "supervisor")
graph.add_edge("writer", "supervisor")

app = graph.compile()

# Run
result = await app.ainvoke({"task": "2026 RAG arxitekturalarini tahlil qiling"})

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Research Analyst",
    goal="2026 backend trendlarini chuqur tadqiq qilish",
    backstory="10 yil tajribali tech analyst",
    tools=[search_tool, browser_tool],
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Research natijalari asosida clear article yozish",
    backstory="Tech blog writer",
    verbose=True
)

research_task = Task(
    description="2026 backend architecture trends haqida comprehensive research",
    agent=researcher,
    expected_output="Detailed findings with sources"
)

write_task = Task(
    description="Research asosida 2000 so'zli blog post yozish",
    agent=writer,
    expected_output="Publication-ready blog post",
    context=[research_task]   # dependency
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential
)

result = crew.kickoff()

"""MCP server — har AI client foydalanishi mumkin"""
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.types as types

server = Server("weather-mcp")

# 1. Tool — LLM chaqirishi mumkin
@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="get_weather",
            description="Shahar ob-havosi",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "units": {"type": "string", "enum": ["c", "f"]}
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "get_weather":
        result = await fetch_weather(arguments["city"], arguments.get("units", "c"))
        return [types.TextContent(type="text", text=json.dumps(result))]

# 2. Resource — static ma'lumot (masalan, fayl)
@server.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="weather://stations",
            name="Weather stations catalog",
            mimeType="application/json"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "weather://stations":
        return json.dumps(await get_stations())

# 3. Prompt — reusable template
@server.list_prompts()
async def list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="weather_report",
            description="Ob-havo hisoboti yaratish",
            arguments=[
                types.PromptArgument(name="city", description="Shahar", required=True)
            ]
        )
    ]

# Run stdio transport
if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server
    
    async def main():
        async with stdio_server() as (read, write):
            await server.run(read, write, InitializationOptions(
                server_name="weather-mcp",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            ))
    
    asyncio.run(main())

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/mcp_server.py"]
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://..."]
    },
    "github": {
      "command": "docker",
      "args": ["run", "-i", "--rm", "-e", "GITHUB_TOKEN", "mcp/github"],
      "env": {"GITHUB_TOKEN": "ghp_..."}
    }
  }
}

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe()
async def rag_pipeline(question: str, tenant_id: str):
    # Avtomatik trace — har step log bo'ladi
    chunks = await retrieve(question)
    answer = await generate(question, chunks)
    return answer

@observe(as_type="generation")
async def generate(question: str, chunks: list):
    response = await client.messages.create(
        model="claude-opus-4-7",
        max_tokens=1000,
        messages=[...]
    )
    # Langfuse avtomatik: input, output, tokens, cost, latency
    return response.content[0].text

# User feedback
@observe()
async def rate_answer(trace_id: str, score: int, comment: str):
    langfuse.score(trace_id=trace_id, name="user_rating", value=score, comment=comment)

"""Smart model router — oddiy savol'ga arzon, qiyin'ga qimmat model"""
import instructor
from pydantic import BaseModel

class Difficulty(BaseModel):
    level: Literal["simple", "medium", "complex"]
    reasoning: str

async def route_request(question: str) -> str:
    # Kichik tez model bilan difficulty classify
    diff = await classifier.messages.create(
        model="claude-haiku-4-5-20251001",
        response_model=Difficulty,
        messages=[{"role": "user", "content": 
                   f"Bu savol qanchalik qiyin?\n{question}"}]
    )
    
    if diff.level == "simple":
        # $0.25/1M input — eng arzon
        return await haiku_answer(question)
    elif diff.level == "medium":
        # $3/1M input — balanced
        return await sonnet_answer(question)
    else:
        # $15/1M input — eng aqlli
        return await opus_answer(question)

# Natijalar (1M so'rov):
#   Hamma Opus'da:  $15,000
#   Router bilan:    $3,200 (80% savol simple)
#   Saving: 80%

import re

# 1. Prompt injection detection
INJECTION_PATTERNS = [
    r"ignore\s+(previous|all)\s+(instructions|prompts)",
    r"you\s+are\s+(now|actually)\s+",
    r"system\s*:\s*",
    r"<\s*system\s*>",
    r"jailbreak|DAN|developer mode",
]

def detect_injection(text: str) -> bool:
    text_lower = text.lower()
    return any(re.search(p, text_lower, re.IGNORECASE) for p in INJECTION_PATTERNS)

# 2. PII redaction (pre-LLM)
import presidio_analyzer
import presidio_anonymizer

analyzer = presidio_analyzer.AnalyzerEngine()
anonymizer = presidio_anonymizer.AnonymizerEngine()

def redact_pii(text: str) -> str:
    results = analyzer.analyze(text=text, language="en",
        entities=["CREDIT_CARD", "EMAIL_ADDRESS", "PHONE_NUMBER", "PERSON", "US_SSN"])
    anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
    return anonymized.text

# 3. LLM output validation
from guardrails import Guard
from guardrails.hub import ValidJson, ToxicLanguage, DetectPII

guard = Guard().use_many(
    ValidJson(on_fail="exception"),
    ToxicLanguage(on_fail="filter"),
    DetectPII(on_fail="filter"),
)

async def safe_llm_call(prompt: str):
    # Input check
    if detect_injection(prompt):
        raise SecurityError("Prompt injection detected")
    
    redacted_prompt = redact_pii(prompt)
    response = await llm.generate(redacted_prompt)
    
    # Output validate
    validated = guard.parse(response)
    return validated.validated_output

import string
import time
import random

BASE62 = string.ascii_letters + string.digits  # 62 ta belgi

# Snowflake-inspired ID generator
class SnowflakeIDGenerator:
    EPOCH = 1700000000000          # ms, custom epoch
    DATACENTER_BITS = 5
    MACHINE_BITS = 5
    SEQUENCE_BITS = 12

    def __init__(self, datacenter_id: int, machine_id: int):
        self.datacenter_id = datacenter_id
        self.machine_id = machine_id
        self.sequence = 0
        self.last_timestamp = -1

    def next_id(self) -> int:
        ts = int(time.time() * 1000) - self.EPOCH
        if ts == self.last_timestamp:
            self.sequence = (self.sequence + 1) & 4095  # 12 bit
            if self.sequence == 0:
                while ts <= self.last_timestamp:
                    ts = int(time.time() * 1000) - self.EPOCH
        else:
            self.sequence = 0
        self.last_timestamp = ts

        # 41 bit ts | 5 bit datacenter | 5 bit machine | 12 bit seq
        return (ts << 22) | (self.datacenter_id << 17) | (self.machine_id << 12) | self.sequence

def encode_base62(num: int) -> str:
    """64-bit int → 7 ta Base62 belgi"""
    chars = []
    while num > 0:
        chars.append(BASE62[num % 62])
        num //= 62
    return ''.join(reversed(chars)).zfill(7)

def decode_base62(code: str) -> int:
    num = 0
    for c in code:
        num = num * 62 + BASE62.index(c)
    return num

# FastAPI endpoint
from fastapi import FastAPI, HTTPException
from fastapi.responses import RedirectResponse
import redis.asyncio as aioredis

app = FastAPI()
redis_client = aioredis.from_url("redis://localhost")
id_gen = SnowflakeIDGenerator(datacenter_id=1, machine_id=1)

@app.post("/shorten")
async def shorten(long_url: str, user_id: str = "anonymous", ttl_days: int = 30):
    # Rate limit check (token bucket from 1.3 bo'limi)
    allowed, remaining = await check_rate_limit(user_id)
    if not allowed:
        raise HTTPException(429, "Rate limit exceeded")

    short_id = encode_base62(id_gen.next_id())

    # Redis'da saqlash (primary fast path)
    await redis_client.setex(
        f"url:{short_id}",
        ttl_days * 86400,
        long_url
    )

    # DB'ga async yozish (Cassandra — eventual consistency OK)
    await db.insert_url(short_id=short_id, long_url=long_url,
                        user_id=user_id, created_at=time.time(),
                        expires_at=time.time() + ttl_days * 86400)

    return {"short_url": f"https://bit.ly/{short_id}", "short_id": short_id}

@app.get("/{short_id}")
async def redirect(short_id: str):
    # 1. Cache'dan qidirish (hit rate ~80%)
    long_url = await redis_client.get(f"url:{short_id}")

    if not long_url:
        # 2. DB'dan qidirish
        record = await db.get_url(short_id)
        if not record or record.expires_at < time.time():
            raise HTTPException(404, "URL not found or expired")
        long_url = record.long_url
        # Cache'ga qo'shish (cache-aside pattern)
        await redis_client.setex(f"url:{short_id}", 86400, long_url)

    # Analytics event async (Kafka)
    await kafka.send("url.clicked", {
        "short_id": short_id, "timestamp": time.time(),
        "user_agent": request.headers.get("user-agent"),
        "ip": request.client.host
    })

    # 301 Permanent (browser caches) yoki 302 Found (har safar serverga)
    # Analytics kerak bo'lsa 302, SEO uchun 301
    return RedirectResponse(url=long_url, status_code=302)

import asyncio
import json
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from datetime import datetime
import uuid

app = FastAPI()

# Connection registry — server ichida
# Production'da Redis Pub/Sub yoki Kafka bilan
class ConnectionManager:
    def __init__(self):
        self.active: dict[str, WebSocket] = {}  # user_id → ws

    async def connect(self, user_id: str, ws: WebSocket):
        await ws.accept()
        self.active[user_id] = ws
        # User online deb belgilash
        await redis.set(f"presence:{user_id}", "online", ex=30)

    def disconnect(self, user_id: str):
        self.active.pop(user_id, None)

    async def send_to(self, user_id: str, message: dict) -> bool:
        ws = self.active.get(user_id)
        if ws:
            try:
                await ws.send_json(message)
                return True
            except Exception:
                self.disconnect(user_id)
        return False

manager = ConnectionManager()

@app.websocket("/ws/{user_id}")
async def websocket_endpoint(ws: WebSocket, user_id: str):
    await manager.connect(user_id, ws)
    try:
        # Offline vaqtdagi xabarlarni yetkazish
        pending = await get_pending_messages(user_id)
        for msg in pending:
            await ws.send_json(msg)
            await mark_delivered(msg["id"])

        while True:
            data = await ws.receive_json()
            await handle_message(user_id, data)
    except WebSocketDisconnect:
        manager.disconnect(user_id)
        await redis.delete(f"presence:{user_id}")

async def handle_message(sender_id: str, data: dict):
    msg = {
        "id": str(uuid.uuid4()),
        "from": sender_id,
        "to": data["to"],
        "text": data["text"],
        "timestamp": datetime.utcnow().isoformat(),
        "status": "sent",
        "client_msg_id": data.get("client_msg_id"),  # dedup uchun
    }

    # 1. Cassandra'ga saqlash (message_history)
    await save_message(msg)

    # 2. Recipient onlinemi?
    is_online = await redis.get(f"presence:{msg['to']}")

    if is_online:
        # 3a. Direct delivery (agar bu serverda)
        delivered = await manager.send_to(msg["to"], msg)
        if not delivered:
            # 3b. Boshqa chat serverda — Kafka orqali
            await kafka.send("messages", msg)
    else:
        # 3c. Push notification
        await push_service.notify(
            user_id=msg["to"],
            title=f"Yangi xabar: {sender_id}",
            body=msg["text"][:50]
        )

CELEBRITY_THRESHOLD = 10_000  # follower soni

async def create_post(user_id: str, content: str) -> dict:
    post = await db.insert_post(user_id=user_id, content=content,
                                created_at=time.time())

    follower_count = await get_follower_count(user_id)

    if follower_count < CELEBRITY_THRESHOLD:
        # Fanout-on-write: barcha follower feed'ini yangilaymiz
        # Async Celery task orqali (HTTP request'ni bloklamaymiz)
        fanout_to_followers.delay(user_id=user_id, post_id=post.id)
    else:
        # Celebrity: faqat post saqlanadi, fan pull qiladi
        # Hech narsa qilmaymiz — read vaqtida merge qilamiz
        pass

    return post

async def get_feed(user_id: str, page: int = 1) -> list:
    # 1. User'ning precomputed feed'ini Redis'dan olamiz
    feed_key = f"feed:{user_id}"
    feed = await redis.lrange(feed_key, (page-1)*20, page*20-1)

    # 2. User follow qilgan celebrity'lar postlarini merge qilamiz
    celebrities = await get_followed_celebrities(user_id)
    celebrity_posts = []
    for celeb_id in celebrities:
        posts = await db.get_recent_posts(user_id=celeb_id, limit=10)
        celebrity_posts.extend(posts)

    # 3. Merge, deduplicate, sort by timestamp
    all_posts = list(feed) + celebrity_posts
    all_posts.sort(key=lambda p: p["created_at"], reverse=True)
    return all_posts[:20]

@celery_app.task
async def fanout_to_followers(user_id: str, post_id: str):
    """Background task: 1000 ta follower uchun ham tez ishlashi kerak"""
    followers = await get_followers_paginated(user_id, batch_size=100)
    pipeline = redis.pipeline()
    for follower_id in followers:
        # Redis list'ga prepend (lpush), max 200 post saqlash (ltrim)
        pipeline.lpush(f"feed:{follower_id}", post_id)
        pipeline.ltrim(f"feed:{follower_id}", 0, 199)
    await pipeline.execute()

import hashlib
from sortedcontainers import SortedDict

class ConsistentHashRing:
    def __init__(self, virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring: SortedDict = SortedDict()
        self.servers: set = set()

    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_server(self, server: str):
        self.servers.add(server)
        for i in range(self.virtual_nodes):
            vnode_key = self._hash(f"{server}:vnode:{i}")
            self.ring[vnode_key] = server

    def remove_server(self, server: str):
        self.servers.discard(server)
        for i in range(self.virtual_nodes):
            vnode_key = self._hash(f"{server}:vnode:{i}")
            self.ring.pop(vnode_key, None)

    def get_server(self, key: str) -> str | None:
        if not self.ring:
            return None
        key_hash = self._hash(key)
        # Soat yo'nalishi bo'yicha keyingi server
        idx = self.ring.bisect_left(key_hash)
        if idx == len(self.ring):
            idx = 0  # ring — aylana, oxiridan boshiga
        return self.ring.peekitem(idx)[1]

    def get_replica_servers(self, key: str, n: int = 3) -> list[str]:
        """N ta unique server (replication uchun)"""
        if not self.ring or len(self.servers) < n:
            return list(self.servers)

        key_hash = self._hash(key)
        idx = self.ring.bisect_left(key_hash)
        seen = set()
        result = []

        for _ in range(len(self.ring)):
            if idx >= len(self.ring):
                idx = 0
            server = self.ring.peekitem(idx)[1]
            if server not in seen:
                seen.add(server)
                result.append(server)
                if len(result) == n:
                    break
            idx += 1
        return result

# Ishlatish:
ring = ConsistentHashRing(virtual_nodes=150)
ring.add_server("cache-1:6379")
ring.add_server("cache-2:6379")
ring.add_server("cache-3:6379")

key = "user:12345:profile"
server = ring.get_server(key)  # "cache-2:6379"
replicas = ring.get_replica_servers(key, n=2)  # ha, replication uchun

"""
GraphRAG pipeline: Document → Entity extraction → Graph → Hybrid retrieval
"""
from anthropic import AsyncAnthropic
from neo4j import AsyncGraphDatabase
import asyncio

client = AsyncAnthropic()

# ── 1. INDEXING: Entity va munosabatlarni chiqarib olish ──
EXTRACTION_PROMPT = """
Quyidagi matndan entity va munosabatlarni JSON formatda chiqar.

Format:
{
  "entities": [{"id": "apple", "type": "COMPANY", "name": "Apple Inc.", "props": {...}}],
  "relationships": [{"from": "tim_cook", "to": "apple", "type": "CEO_OF", "props": {...}}]
}

Faqat JSON qaytargin, boshqa narsa yo'q.

Matn: {text}
"""

async def extract_entities(text: str) -> dict:
    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2000,
        messages=[{"role": "user", "content": EXTRACTION_PROMPT.format(text=text)}]
    )
    import json
    return json.loads(response.content[0].text)

async def index_document(doc_text: str, doc_id: str, neo4j_session):
    """Hujjatni graph'ga qo'shish"""
    extracted = await extract_entities(doc_text)

    # Neo4j'ga entity'lar qo'shish
    for entity in extracted["entities"]:
        await neo4j_session.run(
            "MERGE (e:Entity {id: $id}) SET e += $props SET e:$type",
            id=entity["id"], props=entity.get("props", {}), type=entity["type"]
        )

    # Munosabatlar qo'shish
    for rel in extracted["relationships"]:
        await neo4j_session.run(
            """
            MATCH (a:Entity {id: $from_id}), (b:Entity {id: $to_id})
            MERGE (a)-[r:RELATED {type: $rel_type}]->(b)
            SET r += $props
            """,
            from_id=rel["from"], to_id=rel["to"],
            rel_type=rel["type"], props=rel.get("props", {})
        )

# ── 2. QUERY: Hybrid Graph + Vector retrieval ──
async def hybrid_graphrag_query(question: str, neo4j_session, qdrant) -> str:
    # a) Savoldan entity'larni chiqar
    entities_in_q = await extract_entities(question)
    entity_ids = [e["id"] for e in entities_in_q.get("entities", [])]

    # b) Graph traversal — entity'lardan bog'liq tugunlarni topish (2 hop)
    graph_context = []
    if entity_ids:
        result = await neo4j_session.run(
            """
            MATCH (e:Entity)-[r*1..2]-(related:Entity)
            WHERE e.id IN $entity_ids
            RETURN e, r, related LIMIT 50
            """,
            entity_ids=entity_ids
        )
        graph_context = [record.data() async for record in result]

    # c) Vector search — semantic o'xshash chunk'lar
    q_embedding = await get_embedding(question)
    vector_results = await qdrant.search(
        collection_name="documents",
        query_vector=q_embedding,
        limit=5
    )

    # d) Ikkalasini birlashtirish va LLM'ga berish
    context = f"""
Graph Context (Munosabatlar):
{format_graph_context(graph_context)}

Vector Context (O'xshash hujjatlar):
{chr(10).join(r.payload['text'] for r in vector_results)}
"""

    response = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1500,
        system="Siz berilgan context asosida aniq va to'liq javob beradigan yordamchisiz.",
        messages=[{"role": "user", "content": f"Context:\n{context}\n\nSavol: {question}"}]
    )
    return response.content[0].text

"""
Production AI Memory System — 4 xotira turi
"""
from qdrant_client import AsyncQdrantClient
from datetime import datetime
import json

class AgentMemory:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.qdrant = AsyncQdrantClient("localhost", port=6333)
        self.short_term: list = []      # joriy session context
        self.max_short_term = 20        # max 20 ta turn

    # ── Short-term: conversation context ──
    def add_to_short_term(self, role: str, content: str):
        self.short_term.append({"role": role, "content": content})
        # Eski xabarlarni o'chirish (token limit uchun)
        if len(self.short_term) > self.max_short_term:
            # Birinchi ikki (system) ni qoldirib, engeski'larni o'chiramiz
            self.short_term = self.short_term[:2] + self.short_term[-self.max_short_term+2:]

    def get_short_term_context(self) -> list:
        return self.short_term.copy()

    # ── Long-term: vector DB'da saqlash ──
    async def remember(self, content: str, memory_type: str, metadata: dict = {}):
        """Muhim ma'lumotni uzoq muddatli xotiraga yozish"""
        embedding = await get_embedding(content)
        await self.qdrant.upsert(
            collection_name="agent_memory",
            points=[{
                "id": generate_uuid(),
                "vector": embedding,
                "payload": {
                    "agent_id": self.agent_id,
                    "content": content,
                    "type": memory_type,  # "fact", "preference", "skill"
                    "created_at": datetime.utcnow().isoformat(),
                    **metadata
                }
            }]
        )

    async def recall(self, query: str, memory_type: str = None, limit: int = 5) -> list:
        """Semantic qidirish — eng relevant xotiralarni qaytarish"""
        q_embedding = await get_embedding(query)
        filter_condition = {"must": [{"key": "agent_id", "match": {"value": self.agent_id}}]}
        if memory_type:
            filter_condition["must"].append({"key": "type", "match": {"value": memory_type}})

        results = await self.qdrant.search(
            collection_name="agent_memory",
            query_vector=q_embedding,
            query_filter=filter_condition,
            limit=limit,
            with_payload=True
        )
        return [r.payload for r in results]

    # ── Episodic: nima qilganini yozish ──
    async def log_action(self, action: str, result: str, success: bool):
        """Agent qilgan har bir ish yoziladi"""
        await db.insert({
            "agent_id": self.agent_id,
            "action": action,
            "result": result,
            "success": success,
            "timestamp": datetime.utcnow().isoformat()
        })
        # Xatoni uzoq muddatli xotiraga ham yozamiz (keyingi safar qaytarmaslik uchun)
        if not success:
            await self.remember(
                f"Xato: {action} → {result}",
                memory_type="error",
                metadata={"avoid_repeat": True}
            )

    # ── Full context builder ──
    async def build_context(self, current_query: str) -> str:
        """Agent uchun to'liq kontekst yig'ish"""
        # 1. Relevant uzoq muddatli xotiralar
        long_term = await self.recall(current_query, limit=3)
        # 2. Relevant xatolar (bir xil xatoni qaytarmasin)
        errors = await self.recall(current_query, memory_type="error", limit=2)

        context_parts = []
        if long_term:
            context_parts.append("Oldingi tajribalarim:\n" +
                "\n".join(f"- {m['content']}" for m in long_term))
        if errors:
            context_parts.append("Ilgari qilgan xatolar (takrorlamang):\n" +
                "\n".join(f"- {e['content']}" for e in errors))

        return "\n\n".join(context_parts) if context_parts else ""

# Ishlatish misoli:
memory = AgentMemory(agent_id="agent-001")

# Yangi ma'lumot o'rganildi
await memory.remember(
    "Foydalanuvchi Python'ni afzal ko'radi, JavaScript'dan qochadi",
    memory_type="preference"
)

# Vazifa bajarildi
await memory.log_action(
    action="Write unit tests for auth module",
    result="12 ta test yozildi, 11 ta pass, 1 ta fail (JWT expiry edge case)",
    success=False
)

# Yangi savol keldi — contextni yig'amiz
context = await memory.build_context("Auth module'da yangi feature qo'shish kerak")
# context ichida: preference (Python), xato (JWT expiry) bor

from anthropic import AsyncAnthropic
from openai import AsyncOpenAI
import re

client = AsyncAnthropic()
openai_client = AsyncOpenAI()

DIFFICULTY_CLASSIFIER_PROMPT = """
Quyidagi savolning murakkabligini baholang. Faqat JSON qaytaring:
{
  "difficulty": "easy" | "medium" | "hard",
  "reasoning": "bir jumlada sabab",
  "requires_code": true/false,
  "requires_reasoning": true/false
}

Easy: oddiy savol, faktual, qisqa javob
Medium: tushuntirish kerak, bir necha qadam
Hard: kompleks reasoning, katta kontekst, kod tahlil, multi-step

Savol: {question}
"""

async def classify_difficulty(question: str) -> dict:
    """Kichik model bilan savolni baholash"""
    import json
    response = await openai_client.chat.completions.create(
        model="gpt-4o-mini",   # arzon classifier
        messages=[{"role": "user",
                   "content": DIFFICULTY_CLASSIFIER_PROMPT.format(question=question)}],
        max_tokens=100,
        temperature=0
    )
    return json.loads(response.choices[0].message.content)

async def smart_route(question: str, context: str = "") -> tuple[str, str]:
    """
    Returns: (answer, model_used)
    """
    classification = await classify_difficulty(question)
    difficulty = classification["difficulty"]
    needs_reasoning = classification.get("requires_reasoning", False)
    needs_code = classification.get("requires_code", False)

    # Routing mantiq
    if difficulty == "easy" and not needs_reasoning:
        # Eng arzon: GPT-4o mini ($0.15/1M tokens)
        model = "gpt-4o-mini"
        response = await openai_client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": question}],
            max_tokens=500
        )
        answer = response.choices[0].message.content

    elif difficulty == "medium" or needs_code:
        # O'rta narx: Claude Sonnet ($3/1M tokens)
        model = "claude-sonnet-4-6"
        response = await client.messages.create(
            model=model,
            max_tokens=1500,
            messages=[{"role": "user", "content": question}]
        )
        answer = response.content[0].text

    else:  # hard yoki complex reasoning
        # Eng qimmat: Claude Opus ($15/1M tokens) — faqat kerak bo'lganda
        model = "claude-opus-4-6"
        response = await client.messages.create(
            model=model,
            max_tokens=4000,
            system="Siz ekspert muhandissiz. Kompleks muammolarni chuqur tahlil qiling.",
            messages=[{"role": "user", "content": f"{context}\n\n{question}" if context else question}]
        )
        answer = response.content[0].text

    # Cost tracking (Langfuse yoki custom)
    await track_usage(model=model, tokens_used=estimate_tokens(question + answer),
                      difficulty=difficulty)

    return answer, model

# Natijalar:
# 80% savollar → gpt-4o-mini (0.15$/1M)
# 15% → claude-sonnet (3$/1M)
# 5% → claude-opus (15$/1M)
# O'rtacha narx: ~0.6$/1M (opus-only vs 15x tejash)

from functools import lru_cache
from threading import Lock

# ── Python'da eng sodda Singleton — module import ──
# config.py
class _Config:
    def __init__(self):
        self.db_url = "postgres://..."
        self.debug = False

_instance = _Config()  # Module import'da bir marta yaratiladi

def get_config() -> _Config:
    return _instance  # Har doim bir xil obyekt

# ── Thread-safe Singleton (generic) ──
class Singleton:
    _instance = None
    _lock: Lock = Lock()

    def __new__(cls):
        with cls._lock:
            if not cls._instance:
                cls._instance = super().__new__(cls)
        return cls._instance

# ── FastAPI'da: lru_cache bilan ──
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    db_url: str
    api_key: str

@lru_cache          # Bir marta ishlaydi, natijani cache'laydi = Singleton
def get_settings() -> Settings:
    return Settings()

# ── Singleton anti-pattern: global state testing'ni qiyinlashtiradi ──
# Testing uchun yaxshiroq: Dependency Injection
# class MyService:
#     def __init__(self, config: Config):  # inject
#         self.config = config

from abc import ABC, abstractmethod

# ── Abstract Interface ──
class PaymentProvider(ABC):
    @abstractmethod
    async def charge(self, amount: float, currency: str, card_token: str) -> dict:
        ...

    @abstractmethod
    async def refund(self, payment_id: str, amount: float) -> dict:
        ...

# ── Concrete implementations ──
class StripeProvider(PaymentProvider):
    def __init__(self, api_key: str):
        self.stripe = Stripe(api_key)

    async def charge(self, amount, currency, card_token):
        return await self.stripe.payment_intents.create(
            amount=int(amount * 100),
            currency=currency,
            payment_method=card_token,
        )

    async def refund(self, payment_id, amount):
        return await self.stripe.refunds.create(payment_intent=payment_id)

class PaymeProvider(PaymentProvider):
    async def charge(self, amount, currency, card_token):
        # Payme API logikasi
        ...

    async def refund(self, payment_id, amount):
        ...

# ── Factory ──
class PaymentFactory:
    _providers: dict[str, type[PaymentProvider]] = {
        "stripe": StripeProvider,
        "payme": PaymeProvider,
    }

    @classmethod
    def create(cls, provider_name: str, **kwargs) -> PaymentProvider:
        if provider_name not in cls._providers:
            raise ValueError(f"Unknown provider: {provider_name}. "
                           f"Available: {list(cls._providers)}")
        return cls._providers[provider_name](**kwargs)

    @classmethod
    def register(cls, name: str, provider_class: type[PaymentProvider]):
        """Yangi provider qo'shish (Open/Closed principle)"""
        cls._providers[name] = provider_class

# ── Ishlatish ──
provider = PaymentFactory.create("stripe", api_key=settings.STRIPE_KEY)
result = await provider.charge(99.99, "USD", card_token)

# ── Konfiguratsiyadan ──
provider_name = settings.PAYMENT_PROVIDER  # "stripe" yoki "payme"
provider = PaymentFactory.create(provider_name, **settings.payment_config)

from dataclasses import dataclass, field
from typing import Self

# ── SQL Query Builder ──
class QueryBuilder:
    """Method chaining bilan SQL query yaratish"""

    def __init__(self, table: str):
        self._table = table
        self._conditions: list[str] = []
        self._columns: list[str] = ["*"]
        self._limit: int | None = None
        self._offset: int = 0
        self._order: str | None = None
        self._params: list = []

    def select(self, *columns: str) -> Self:
        self._columns = list(columns)
        return self

    def where(self, condition: str, *values) -> Self:
        param_start = len(self._params) + 1
        self._conditions.append(
            condition.replace("?", f"${param_start}")
        )
        self._params.extend(values)
        return self

    def limit(self, n: int) -> Self:
        self._limit = n
        return self

    def offset(self, n: int) -> Self:
        self._offset = n
        return self

    def order_by(self, column: str, direction: str = "ASC") -> Self:
        self._order = f"{column} {direction}"
        return self

    def build(self) -> tuple[str, list]:
        cols = ", ".join(self._columns)
        sql = f"SELECT {cols} FROM {self._table}"
        if self._conditions:
            sql += " WHERE " + " AND ".join(self._conditions)
        if self._order:
            sql += f" ORDER BY {self._order}"
        if self._limit:
            sql += f" LIMIT {self._limit}"
        if self._offset:
            sql += f" OFFSET {self._offset}"
        return sql, self._params

# ── Ishlatish ──
query, params = (
    QueryBuilder("orders")
    .select("id", "status", "total")
    .where("tenant_id = ?", tenant_id)
    .where("status = ?", "active")
    .order_by("created_at", "DESC")
    .limit(20)
    .offset(40)
    .build()
)
# SELECT id, status, total FROM orders
# WHERE tenant_id = $1 AND status = $2
# ORDER BY created_at DESC LIMIT 20 OFFSET 40

from abc import ABC, abstractmethod

# ── Abstract interface ──
class OrderRepository(ABC):
    @abstractmethod
    async def get_by_id(self, order_id: str) -> Order | None: ...
    @abstractmethod
    async def get_by_tenant(self, tenant_id: str, limit: int, offset: int) -> list[Order]: ...
    @abstractmethod
    async def create(self, data: OrderCreate, user_id: str) -> Order: ...
    @abstractmethod
    async def update_status(self, order_id: str, status: str) -> Order: ...

# ── PostgreSQL implementatsiyasi ──
class PostgresOrderRepository(OrderRepository):
    def __init__(self, db):
        self.db = db

    async def get_by_id(self, order_id: str) -> Order | None:
        row = await self.db.fetchrow(
            "SELECT * FROM orders WHERE id = $1", order_id
        )
        return Order(**dict(row)) if row else None

    async def get_by_tenant(self, tenant_id, limit, offset) -> list[Order]:
        rows = await self.db.fetch(
            """SELECT * FROM orders WHERE tenant_id = $1
               ORDER BY created_at DESC LIMIT $2 OFFSET $3""",
            tenant_id, limit, offset
        )
        return [Order(**dict(r)) for r in rows]

    async def create(self, data: OrderCreate, user_id: str) -> Order:
        row = await self.db.fetchrow(
            """INSERT INTO orders (id, user_id, product_id, quantity, status)
               VALUES (gen_random_uuid(), $1, $2, $3, 'pending')
               RETURNING *""",
            user_id, data.product_id, data.quantity
        )
        return Order(**dict(row))

    async def update_status(self, order_id: str, status: str) -> Order:
        row = await self.db.fetchrow(
            "UPDATE orders SET status=$1, updated_at=NOW() WHERE id=$2 RETURNING *",
            status, order_id
        )
        return Order(**dict(row))

# ── In-memory implementatsiyasi (testing uchun!) ──
class InMemoryOrderRepository(OrderRepository):
    def __init__(self):
        self._store: dict[str, Order] = {}

    async def get_by_id(self, order_id: str) -> Order | None:
        return self._store.get(order_id)

    async def create(self, data: OrderCreate, user_id: str) -> Order:
        order = Order(id=str(uuid.uuid4()), user_id=user_id, **data.dict())
        self._store[order.id] = order
        return order

# ── Service — Repository inject qilinadi ──
class OrderService:
    def __init__(self, repo: OrderRepository):  # Abstract type!
        self.repo = repo

    async def create_order(self, data: OrderCreate, user: User) -> Order:
        # Biznes logikasi faqat shu yerda
        if data.quantity > 100:
            raise ValueError("Maximum 100 ta buyurtma")
        order = await self.repo.create(data, user.id)
        await self.notify_warehouse(order)
        return order

# Dependency injection:
# Production: OrderService(PostgresOrderRepository(db))
# Test:       OrderService(InMemoryOrderRepository())

from abc import ABC, abstractmethod

# ── Notification strategy ──
class NotificationStrategy(ABC):
    @abstractmethod
    async def send(self, user: User, message: str) -> bool: ...

class EmailStrategy(NotificationStrategy):
    async def send(self, user, message):
        await sendgrid.send(to=user.email, body=message)
        return True

class SMSStrategy(NotificationStrategy):
    async def send(self, user, message):
        await twilio.send(to=user.phone, body=message[:160])
        return True

class PushStrategy(NotificationStrategy):
    async def send(self, user, message):
        await fcm.send(token=user.device_token, body=message)
        return True

class MultiStrategy(NotificationStrategy):
    """Bir nechta kanal — fallback bilan"""
    def __init__(self, strategies: list[NotificationStrategy]):
        self.strategies = strategies

    async def send(self, user, message):
        for strategy in self.strategies:
            try:
                success = await strategy.send(user, message)
                if success:
                    return True
            except Exception:
                continue  # Keyingi kanalga o'tish
        return False

# ── Notification service — strategy inject ──
class NotificationService:
    def __init__(self, strategy: NotificationStrategy):
        self.strategy = strategy

    async def notify(self, user: User, message: str):
        return await self.strategy.send(user, message)

    def set_strategy(self, strategy: NotificationStrategy):
        self.strategy = strategy  # Runtime'da almashtirish

# ── Konfiguratsiyadan strategy tanlash ──
STRATEGIES = {
    "email": EmailStrategy(),
    "sms": SMSStrategy(),
    "push": PushStrategy(),
    "multi": MultiStrategy([PushStrategy(), EmailStrategy()]),
}

notifier = NotificationService(STRATEGIES[settings.NOTIFICATION_CHANNEL])

import asyncio
from collections import defaultdict

# ── Simple async event bus ──
class EventBus:
    _instance = None
    _handlers: dict[str, list] = defaultdict(list)

    @classmethod
    def subscribe(cls, event: str, handler):
        """Event'ga handler qo'shish"""
        cls._handlers[event].append(handler)

    @classmethod
    async def publish(cls, event: str, data: dict):
        """Event chiqarish — barcha handler'larni chaqirish"""
        handlers = cls._handlers.get(event, [])
        if not handlers:
            return
        # Barcha handler'lar parallel ishlaydi
        await asyncio.gather(
            *[handler(data) for handler in handlers],
            return_exceptions=True  # Bir handler xatosi boshqasini to'xtatmasin
        )

# ── Handler'lar ──
async def send_order_confirmation(data: dict):
    await email_service.send(
        to=data["email"],
        subject=f"Order #{data['order_id']} confirmed",
        template="order_confirmation"
    )

async def update_inventory(data: dict):
    await inventory_service.decrease(
        product_id=data["product_id"],
        quantity=data["quantity"]
    )

async def notify_warehouse(data: dict):
    await warehouse_api.post("/new-order", json=data)

# ── Subscribe ──
EventBus.subscribe("order.created", send_order_confirmation)
EventBus.subscribe("order.created", update_inventory)
EventBus.subscribe("order.created", notify_warehouse)

# ── Publish ──
@router.post("/orders")
async def create_order(data: OrderCreate, user: User = Depends(get_current_user)):
    order = await order_service.create(data, user)

    # Event chiqarish — handler'lar parallel ishlaydi
    await EventBus.publish("order.created", {
        "order_id": str(order.id),
        "email": user.email,
        "product_id": str(data.product_id),
        "quantity": data.quantity,
    })

    return order

"""
Microservices o'rtasida kommunikatsiya:
1. Sync REST/gRPC — darhol javob kerak
2. Async Kafka — eventual consistency OK
3. Service mesh — mTLS, retry, circuit breaker
"""

# ── 1. Sync gRPC — internal service call ──
import grpc
from proto import orders_pb2, orders_pb2_grpc

async def get_order_status(order_id: str) -> str:
    """Sync call — javob kerak"""
    async with grpc.aio.insecure_channel("orders-service:50051") as channel:
        stub = orders_pb2_grpc.OrdersStub(channel)
        response = await stub.GetOrder(
            orders_pb2.GetOrderRequest(order_id=order_id),
            timeout=2.0  # 2 sekund timeout
        )
        return response.status

# ── 2. Circuit Breaker ──
from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=30)
async def call_payment_service(order_id: str, amount: float):
    """5 marta fail bo'lsa — 30 sek "open" holatda"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://payment-service/charge",
            json={"order_id": order_id, "amount": amount},
            timeout=5.0
        )
        response.raise_for_status()
        return response.json()

# ── 3. Async via Kafka ──
async def process_order_completed(order: Order):
    """Order tugagach — async fan-out"""
    # Payment service'ni async xabardor qilish
    await kafka.send("order.completed", {
        "order_id": str(order.id),
        "user_id": str(order.user_id),
        "amount": float(order.total),
        "items": [{"product_id": str(i.product_id), "qty": i.quantity}
                  for i in order.items]
    })
    # Har service o'z tezligida o'qiydi:
    # - notification-service: email yuboradi
    # - inventory-service: stokni kamaytiradi
    # - analytics-service: statistika yangilaydi

# ── 4. Saga — distributed transaction ──
class OrderSaga:
    """Kompensatsiya bilan distributed transaction"""
    steps = []
    compensations = []

    async def execute(self, order_data):
        try:
            # Step 1: Payment
            payment = await payment_service.charge(order_data)
            self.compensations.append(
                lambda: payment_service.refund(payment.id)
            )

            # Step 2: Inventory
            reservation = await inventory_service.reserve(order_data)
            self.compensations.append(
                lambda: inventory_service.release(reservation.id)
            )

            # Step 3: Order create
            order = await order_service.create(order_data, payment.id)
            return order

        except Exception as e:
            # Ro'y bergan qadam'gacha kompensatsiya
            for compensate in reversed(self.compensations):
                await compensate()
            raise

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

def setup_tracing(app: FastAPI, service_name: str):
    """OpenTelemetry — auto-instrumentation + manual spans"""

    # Exporter — Jaeger yoki Tempo'ga yuborish
    exporter = OTLPSpanExporter(endpoint="http://jaeger:4317")
    provider = TracerProvider()
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)

    # Auto-instrumentation — FastAPI, asyncpg, httpx
    FastAPIInstrumentor.instrument_app(app)
    AsyncPGInstrumentor().instrument()
    HTTPXClientInstrumentor().instrument()

# ── Manual span — muhim funksiya uchun ──
tracer = trace.get_tracer(__name__)

async def process_order(order_id: str):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", order_id)

        # DB span avtomatik qo'shiladi (asyncpg instrumentation)
        order = await db.fetchrow("SELECT * FROM orders WHERE id = $1", order_id)

        with tracer.start_as_current_span("validate_inventory"):
            is_available = await check_inventory(order["product_id"])
            span.set_attribute("inventory.available", is_available)

        if not is_available:
            span.set_attribute("order.cancelled", True)
            span.set_status(trace.Status(trace.StatusCode.ERROR))
            raise OutOfStockError(order["product_id"])

        return order

# Jaeger'da ko'rish:
# /api/v1/orders/create → [5ms]
#   → Auth middleware [1ms]
#   → process_order [12ms]
#     → DB query [4ms]
#     → validate_inventory [6ms]
#       → inventory-service HTTP call [5ms]