# REST endpoint with versioning and RFC 9457 errors
from fastapi import APIRouter, Depends, Request
from fastapi.responses import JSONResponse

router = APIRouter()

@router.get("/api/v1/users/{user_id}")
async def get_user(user_id: str, service: UserService = Depends()):
    user = await service.get_user(user_id)
    if not user:
        raise NotFoundProblem(
            resource="User",
            resource_id=user_id,
        )
    return UserResponseV1(id=user.id, name=user.full_name)

# app/core/problem_types.py
PROBLEM_TYPES = {
    "validation-error": {
        "uri": "https://api.example.com/problems/validation-error",
        "title": "Validation Error",
        "status": 422,
    },
    "resource-not-found": {
        "uri": "https://api.example.com/problems/resource-not-found",
        "title": "Resource Not Found",
        "status": 404,
    },
    "rate-limit-exceeded": {
        "uri": "https://api.example.com/problems/rate-limit-exceeded",
        "title": "Too Many Requests",
        "status": 429,
    },
    "unauthorized": {
        "uri": "https://api.example.com/problems/unauthorized",
        "title": "Unauthorized",
        "status": 401,
    },
    "forbidden": {
        "uri": "https://api.example.com/problems/forbidden",
        "title": "Forbidden",
        "status": 403,
    },
    "conflict": {
        "uri": "https://api.example.com/problems/conflict",
        "title": "Conflict",
        "status": 409,
    },
    "internal-error": {
        "uri": "https://api.example.com/problems/internal-error",
        "title": "Internal Server Error",
        "status": 500,
    },
}

class NotFoundProblem(ProblemException):
    def __init__(self, resource: str, resource_id: str, instance: str | None = None):
        super().__init__(
            status_code=404,
            problem_type="https://api.example.com/problems/resource-not-found",
            title="Resource Not Found",
            detail=f"{resource} with ID '{resource_id}' was not found",
            instance=instance,
            resource=resource,
            resource_id=resource_id,
        )

class RateLimitProblem(ProblemException):
    def __init__(self, retry_after: int, instance: str | None = None):
        super().__init__(
            status_code=429,
            problem_type="https://api.example.com/problems/rate-limit-exceeded",
            title="Too Many Requests",
            detail="Rate limit exceeded. Please retry later.",
            instance=instance,
            retry_after=retry_after,
        )

class ConflictProblem(ProblemException):
    def __init__(self, detail: str, conflicting_field: str | None = None):
        super().__init__(
            status_code=409,
            problem_type="https://api.example.com/problems/conflict",
            title="Resource Conflict",
            detail=detail,
            conflicting_field=conflicting_field,
        )

@router.get("/api/v1/analyses/{analysis_id}")
async def get_analysis(
    analysis_id: str,
    request: Request,
    service: AnalysisService = Depends(get_analysis_service),
):
    analysis = await service.get_by_id(analysis_id)
    if not analysis:
        raise NotFoundProblem(
            resource="Analysis",
            resource_id=analysis_id,
            instance=str(request.url),
        )
    return analysis

import httpx
from dataclasses import dataclass

@dataclass
class ProblemDetail:
    type: str
    status: int
    title: str | None = None
    detail: str | None = None
    instance: str | None = None
    extensions: dict | None = None

    @classmethod
    def from_response(cls, response: httpx.Response) -> "ProblemDetail":
        if response.headers.get("content-type", "").startswith(
            "application/problem+json"
        ):
            data = response.json()
            return cls(
                type=data.get("type", "about:blank"),
                status=data.get("status", response.status_code),
                title=data.get("title"),
                detail=data.get("detail"),
                instance=data.get("instance"),
                extensions={
                    k: v for k, v in data.items()
                    if k not in ("type", "status", "title", "detail", "instance")
                },
            )
        return cls(type="about:blank", status=response.status_code)

interface ProblemDetail {
  type: string;
  status: number;
  title?: string;
  detail?: string;
  instance?: string;
  [key: string]: unknown; // Extensions
}

class APIError extends Error {
  constructor(public problem: ProblemDetail) {
    super(problem.detail || problem.title || 'Unknown error');
  }
}

async function fetchWithProblemDetails(url: string): Promise<Response> {
  const response = await fetch(url);

  if (!response.ok) {
    const contentType = response.headers.get('content-type');

    if (contentType?.includes('application/problem+json')) {
      const problem: ProblemDetail = await response.json();
      throw new APIError(problem);
    }

    throw new APIError({
      type: 'about:blank',
      status: response.status,
      title: response.statusText,
    });
  }

  return response;
}

# Different format every endpoint
raise HTTPException(404, detail="Not found")
raise HTTPException(404, detail={"error": "Missing"})
return JSONResponse({"message": "No such resource"}, 404)

# Consistent use of problem types
raise NotFoundProblem(
    resource="Analysis",
    resource_id=analysis_id,
    instance=str(request.url)
)
# Always returns: application/problem+json with standard fields

from pydantic import BaseModel, Field, HttpUrl
from typing import Any

class ProblemDetail(BaseModel):
    """RFC 9457 Problem Details for HTTP APIs."""

    type: HttpUrl = Field(
        default="about:blank",
        description="URI identifying the problem type",
    )
    title: str = Field(
        description="Short, human-readable summary",
    )
    status: int = Field(
        ge=400, le=599,
        description="HTTP status code",
    )
    detail: str | None = Field(
        default=None,
        description="Human-readable explanation specific to this occurrence",
    )
    instance: str | None = Field(
        default=None,
        description="URI reference identifying the specific occurrence",
    )

    model_config = {"extra": "allow"}  # Allow extension fields

from fastapi import HTTPException
from typing import Any

class ProblemException(HTTPException):
    """Base exception for RFC 9457 problem details."""

    def __init__(
        self,
        status_code: int,
        problem_type: str,
        title: str,
        detail: str | None = None,
        instance: str | None = None,
        **extensions: Any,
    ):
        self.problem_type = problem_type
        self.title = title
        self.detail = detail
        self.instance = instance
        self.extensions = extensions
        super().__init__(status_code=status_code, detail=detail)

    def to_problem_detail(self) -> dict[str, Any]:
        result = {
            "type": self.problem_type,
            "title": self.title,
            "status": self.status_code,
        }
        if self.detail:
            result["detail"] = self.detail
        if self.instance:
            result["instance"] = self.instance
        result.update(self.extensions)
        return result

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

app = FastAPI()

@app.exception_handler(ProblemException)
async def problem_exception_handler(request: Request, exc: ProblemException):
    return JSONResponse(
        status_code=exc.status_code,
        content=exc.to_problem_detail(),
        media_type="application/problem+json",
    )

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={
            "type": "https://api.example.com/problems/internal-error",
            "title": "Internal Server Error",
            "status": 500,
            "detail": "An unexpected error occurred",
            "instance": str(request.url),
        },
        media_type="application/problem+json",
    )

{
  "type": "https://api.example.com/problems/resource-not-found",
  "title": "Resource Not Found",
  "status": 404,
  "detail": "Analysis with ID 'abc123' was not found",
  "instance": "/api/v1/analyses/abc123",
  "resource": "Analysis",
  "resource_id": "abc123"
}

{
  "type": "https://api.example.com/problems/rate-limit-exceeded",
  "title": "Too Many Requests",
  "status": 429,
  "detail": "Rate limit exceeded. Please retry later.",
  "instance": "/api/v1/analyses",
  "retry_after": 60
}

# NEVER return plain text errors
return Response("Not found", status_code=404)

# NEVER use inconsistent error formats
return {"error": "Not found"}           # Different from other errors
return {"message": "Validation failed"}  # Yet another format

# NEVER expose internal details in production
return {"detail": str(exc), "traceback": traceback.format_exc()}

# Different structure every time
return {"error": "Not found"}
return {"message": "Validation failed", "fields": [...]}
return Response("Internal error", 500)

# Consistent RFC 9457 format
return JSONResponse(
    content={
        "type": "https://api.example.com/problems/validation-error",
        "title": "Validation Error",
        "status": 422,
        "errors": [{"field": "url", "message": "Invalid format"}]
    },
    media_type="application/problem+json"
)

class ValidationProblem(ProblemException):
    def __init__(self, errors: list[dict], instance: str | None = None):
        super().__init__(
            status_code=422,
            problem_type="https://api.example.com/problems/validation-error",
            title="Validation Error",
            detail="One or more fields failed validation",
            instance=instance,
            errors=errors,  # Extension field
        )

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
    request: Request, exc: RequestValidationError
):
    errors = [
        {
            "field": ".".join(str(loc) for loc in err["loc"][1:]),
            "code": err["type"],
            "message": err["msg"],
        }
        for err in exc.errors()
    ]
    problem = ValidationProblem(errors=errors, instance=str(request.url))
    return JSONResponse(
        status_code=422,
        content=problem.to_problem_detail(),
        media_type="application/problem+json",
    )

{
  "type": "https://api.example.com/problems/validation-error",
  "title": "Validation Error",
  "status": 422,
  "detail": "One or more fields failed validation",
  "instance": "/api/v1/analyses",
  "errors": [
    {"field": "source_url", "code": "url_parsing", "message": "Invalid URL format"},
    {"field": "depth", "code": "less_than_equal", "message": "Must be between 1 and 3"}
  ]
}

@router.post("/analyses")
async def create_analysis(
    request: AnalyzeRequest,
    service: AnalysisService = Depends(),
):
    # Custom validation beyond Pydantic schema
    if not is_valid_url(str(request.url)):
        raise ValidationProblem(
            errors=[
                {
                    "field": "url",
                    "code": "invalid_url",
                    "message": "URL is not accessible or returns an error",
                }
            ]
        )
    return await service.create(request)

type CreateUserPayload {
  user: User
  errors: [UserError!]
}

type UserError {
  field: String!
  message: String!
  code: String!
}

{
  "data": {
    "createUser": {
      "user": null,
      "errors": [
        {
          "field": "email",
          "message": "Email is already taken",
          "code": "DUPLICATE_EMAIL"
        }
      ]
    }
  }
}

@pytest.mark.asyncio
async def test_validation_error_includes_field_errors(client: AsyncClient):
    response = await client.post("/api/v1/analyses", json={"url": "not-a-url"})

    assert response.status_code == 422
    assert response.headers["content-type"] == "application/problem+json"

    problem = response.json()
    assert problem["type"].endswith("validation-error")
    assert "errors" in problem
    assert any(e["field"] == "url" for e in problem["errors"])

# No field-level details
return {"error": "Validation failed"}, 422

# Specific field errors
return {
    "type": "https://api.example.com/problems/validation-error",
    "status": 422,
    "errors": [
        {"field": "url", "code": "url_parsing", "message": "Invalid URL format"},
        {"field": "depth", "code": "less_than_equal", "message": "Must be between 1 and 3"}
    ]
}, 422

openapi: 3.1.0

info:
  title: Your API Name
  version: 1.0.0
  description: |
    Brief description of what this API does.

    ## Authentication
    This API uses Bearer tokens for authentication.

    ## Rate Limiting
    - 1000 requests per hour per API key

servers:
  - url: https://api.company.com/v1
    description: Production server
  - url: http://localhost:3000/v1
    description: Local development

@router.get(
    "/analyses/{analysis_id}",
    responses={
        404: {"model": ErrorResponse, "description": "Analysis not found"},
        500: {"model": ErrorResponse, "description": "Internal server error"},
    },
    summary="Get analysis details",
    description="Retrieve detailed information about a specific analysis",
)
async def get_analysis(
    analysis_id: Annotated[uuid.UUID, Path(description="Analysis UUID")]
) -> AnalysisResponse:
    ...

components:
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT

  parameters:
    PageParam:
      name: page
      in: query
      schema:
        type: integer
        minimum: 1
        default: 1

    PerPageParam:
      name: per_page
      in: query
      schema:
        type: integer
        minimum: 1
        maximum: 100
        default: 20

  headers:
    X-RateLimit-Limit:
      description: Maximum requests allowed per hour
      schema:
        type: integer
        example: 1000

    X-RateLimit-Remaining:
      description: Requests remaining in current window
      schema:
        type: integer

  schemas:
    Error:
      type: object
      required: [error]
      properties:
        error:
          type: object
          required: [code, message]
          properties:
            code:
              type: string
              example: "VALIDATION_ERROR"
            message:
              type: string
            details:
              type: array
              items:
                type: object
                properties:
                  field:
                    type: string
                  message:
                    type: string
            request_id:
              type: string

  responses:
    NotFoundError:
      description: Resource not found
      content:
        application/json:
          schema:
            $ref: '#/components/schemas/Error'

    ValidationError:
      description: Validation failed
      content:
        application/json:
          schema:
            $ref: '#/components/schemas/Error'

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

def custom_openapi_v1():
    return get_openapi(
        title="OrchestKit API",
        version="1.0.0",
        routes=v1_router.routes,
    )

def custom_openapi_v2():
    return get_openapi(
        title="OrchestKit API",
        version="2.0.0",
        routes=v2_router.routes,
    )

app.mount("/docs/v1", create_docs_app(custom_openapi_v1))
app.mount("/docs/v2", create_docs_app(custom_openapi_v2))

from pydantic import BaseModel, HttpUrl, Field

class AnalyzeRequest(BaseModel):
    url: HttpUrl
    analysis_id: str | None = None
    skill_level: str = Field(
        default="beginner",
        pattern="^(beginner|intermediate|advanced)$",
    )

# No response schema or error docs
@router.get("/analyses/{id}")
async def get_analysis(id: str):
    return await service.get(id)

@router.get(
    "/analyses/{id}",
    responses={
        404: {"model": ErrorResponse, "description": "Analysis not found"},
        500: {"model": ErrorResponse, "description": "Internal error"}
    },
    summary="Get analysis details"
)
async def get_analysis(id: Annotated[str, Path(description="Analysis UUID")]) -> AnalysisResponse:
    return await service.get(id)

# Express ownership and containment through URL hierarchy
GET /api/v1/analyses/{analysis_id}/artifact
GET /api/v1/teams/{team_id}/members
POST /api/v1/projects/{project_id}/tasks

# NOT query params for primary relationships
GET /api/v1/artifact?analysis_id={id}      # Avoid
GET /api/v1/analysis_artifact/{id}          # Avoid

@router.get("/analyses")
async def list_analyses(
    status: str | None = None,
    content_type: str | None = None,
    created_after: datetime | None = None,
    created_before: datetime | None = None,
) -> list[AnalysisResponse]:
    filters = {}
    if status:
        filters["status"] = status
    if content_type:
        filters["content_type"] = content_type
    return await repo.find_all(filters=filters)

GET /api/v1/analyses?status=completed&content_type=article
GET /api/v1/analyses?created_after=2025-01-01&created_before=2025-12-31

@router.get("/analyses")
async def list_analyses(
    sort: str = Query(default="-created_at"),
) -> list[AnalysisResponse]:
    direction = "desc" if sort.startswith("-") else "asc"
    field = sort.lstrip("-")
    return await repo.find_all(order_by=field, direction=direction)

GET /api/v1/analyses?sort=-created_at       # Newest first
GET /api/v1/analyses?sort=title             # Alphabetical

@router.get("/analyses")
async def list_analyses(
    fields: str | None = None,
) -> list[dict[str, Any]]:
    selected_fields = fields.split(",") if fields else None
    results = await repo.find_all()

    if selected_fields:
        return [
            {k: v for k, v in item.dict().items() if k in selected_fields}
            for item in results
        ]
    return results

GET /api/v1/analyses?fields=id,title,status

type Query {
  users(first: Int, after: String): UserConnection!
}

type UserConnection {
  edges: [UserEdge!]!
  pageInfo: PageInfo!
  totalCount: Int!
}

type UserEdge {
  node: User!
  cursor: String!
}

type PageInfo {
  hasNextPage: Boolean!
  hasPreviousPage: Boolean!
  startCursor: String
  endCursor: String
}

# Empty collections: Return empty array, not null
{"data": []}   # Correct
{"data": null}  # Wrong

# Deleted resources: Return 404, not null
# 404 Not Found (correct)
# {"data": null} (wrong)

# Null fields: Be explicit
{"title": null, "description": ""}  # Clear intent

// Parent relationship in query param
GET /api/v1/artifacts?analysis_id=abc-123

// Express ownership in URL structure
GET /api/v1/analyses/abc-123/artifacts

# Plural nouns for collections
GET /users
GET /users/123
POST /users

# Hierarchical relationships
GET /users/123/orders          # Orders for specific user
GET /teams/5/members           # Members of specific team
POST /projects/10/tasks        # Create task in project 10

# Kebab-case for multi-word resources
/shopping-carts
/order-items
/user-preferences

@router.get("/analyses")
async def list_analyses(
    cursor: str | None = None,
    limit: int = Query(default=20, le=100)
) -> PaginatedResponse:
    results = await repo.get_paginated(cursor=cursor, limit=limit)

    return {
        "data": results,
        "pagination": {
            "next_cursor": encode_cursor(results[-1].id) if results else None,
            "has_more": len(results) == limit
        }
    }

# RPC-style endpoints
POST /createUser
POST /users/123/delete
GET /getUserOrders?id=123

# Resource-oriented
POST /users
DELETE /users/123
GET /users/123/orders

# No authentication or authorization
@strawberry.type
class Query:
    @strawberry.field
    async def all_users(self, info: strawberry.Info) -> list[User]:
        return await info.context.user_service.list_all()  # Anyone can see all users!

# Exposing internal IDs
@strawberry.type
class User:
    id: int  # Exposes auto-increment ID!

from strawberry.permission import BasePermission

class IsAuthenticated(BasePermission):
    message = "User is not authenticated"

    async def has_permission(self, source, info: strawberry.Info, **kwargs) -> bool:
        return info.context.current_user_id is not None

class IsAdmin(BasePermission):
    message = "Admin access required"

    async def has_permission(self, source, info: strawberry.Info, **kwargs) -> bool:
        user_id = info.context.current_user_id
        if not user_id:
            return False
        user = await info.context.user_service.get(user_id)
        return user and user.role == "admin"

@strawberry.type
class Query:
    @strawberry.field(permission_classes=[IsAuthenticated])
    async def me(self, info: strawberry.Info) -> User:
        return await info.context.user_service.get(info.context.current_user_id)

    @strawberry.field(permission_classes=[IsAdmin])
    async def all_users(self, info: strawberry.Info) -> list[User]:
        return await info.context.user_service.list_all()

from strawberry.fastapi import GraphQLRouter

schema = strawberry.Schema(query=Query, mutation=Mutation, subscription=Subscription)

async def get_context(request: Request, user_service=Depends(get_user_service)) -> GraphQLContext:
    return GraphQLContext(request=request, user_service=user_service)

graphql_router = GraphQLRouter(schema, context_getter=get_context, graphiql=True)
app = FastAPI()
app.include_router(graphql_router, prefix="/graphql")

@strawberry.type
class Subscription:
    @strawberry.subscription
    async def user_updated(self, info: strawberry.Info, user_id: strawberry.ID) -> AsyncGenerator[User, None]:
        async for message in info.context.pubsub.subscribe(f"user:{user_id}:updated"):
            yield User(**message)

# Making database calls in resolver loops
@strawberry.type
class Post:
    author_id: strawberry.ID

    @strawberry.field
    async def author(self, info: strawberry.Info) -> "User":
        # N+1: One query per post!
        return await db.get_user(self.author_id)

from strawberry.dataloader import DataLoader

class UserLoader(DataLoader[str, "User"]):
    def __init__(self, user_repo):
        super().__init__(load_fn=self.batch_load)
        self.user_repo = user_repo

    async def batch_load(self, keys: list[str]) -> list["User"]:
        users = await self.user_repo.get_many(keys)
        user_map = {u.id: u for u in users}
        return [user_map.get(key) for key in keys]

@strawberry.type
class Post:
    author_id: strawberry.ID

    @strawberry.field
    async def author(self, info: strawberry.Info) -> "User":
        return await info.context.user_loader.load(self.author_id)

import strawberry
from strawberry import Private

@strawberry.type
class User:
    id: strawberry.ID
    email: str
    name: str
    password_hash: Private[str]  # Not exposed in schema

    @strawberry.field
    def display_name(self) -> str:
        return f"{self.name} ({self.email})"

@strawberry.input
class CreateUserInput:
    email: str
    name: str
    password: str

@strawberry.type
class CreateUserSuccess:
    user: User

@strawberry.type
class UserError:
    message: str
    code: str
    field: str | None = None

@strawberry.type
class CreateUserError:
    errors: list[UserError]

CreateUserResult = strawberry.union("CreateUserResult", [CreateUserSuccess, CreateUserError])

# Using REST for high-frequency internal calls
response = requests.post("http://user-service/api/users",
                         json={"email": email, "name": name})
# High serialization overhead, no compile-time validation, manual error mapping

syntax = "proto3";
package user.v1;

import "google/protobuf/timestamp.proto";

service UserService {
  rpc GetUser(GetUserRequest) returns (User);
  rpc CreateUser(CreateUserRequest) returns (User);
  rpc ListUsers(ListUsersRequest) returns (stream User);       // Server streaming
  rpc BulkCreateUsers(stream CreateUserRequest) returns (BulkCreateResponse);  // Client streaming
}

message User {
  string id = 1;
  string email = 2;
  string name = 3;
  UserStatus status = 4;
  google.protobuf.Timestamp created_at = 5;
}

enum UserStatus {
  USER_STATUS_UNSPECIFIED = 0;
  USER_STATUS_ACTIVE = 1;
  USER_STATUS_INACTIVE = 2;
}

import grpc.aio
from app.protos import user_service_pb2 as pb2, user_service_pb2_grpc as pb2_grpc

class UserServiceServicer(pb2_grpc.UserServiceServicer):
    async def GetUser(self, request, context):
        user = await self.user_repo.get(request.user_id)
        if not user:
            await context.abort(grpc.StatusCode.NOT_FOUND, f"User {request.user_id} not found")
        return self._to_proto(user)

    async def CreateUser(self, request, context):
        if not request.email or "@" not in request.email:
            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Invalid email")
        user = await self.user_repo.create(email=request.email, name=request.name)
        return self._to_proto(user)

class UserServiceClient:
    def __init__(self, host: str = "localhost:50051"):
        self.channel = grpc.insecure_channel(host, options=[
            ("grpc.keepalive_time_ms", 30000),
            ("grpc.keepalive_timeout_ms", 10000),
        ])
        self.stub = pb2_grpc.UserServiceStub(self.channel)

    def get_user(self, user_id: str, timeout: float = 5.0):
        try:
            return self.stub.GetUser(pb2.GetUserRequest(user_id=user_id), timeout=timeout)
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                raise UserNotFoundError(user_id)
            raise

# Client may disconnect but server keeps processing
def ListUsers(self, request, context):
    for user in all_users:
        yield self._to_proto(user)  # No cancellation check!

def ListUsers(self, request, context):
    """Server streaming: yield users one by one."""
    for user in self.user_repo.iterate(page_size=request.page_size or 100):
        if not context.is_active():  # Check if client disconnected
            return
        yield self._to_proto(user)

async def UserUpdates(self, request_iterator, context):
    """Bidirectional: receive updates, yield results."""
    async for request in request_iterator:
        if not context.is_active():
            return
        user = await self.user_repo.update(request.user_id, request.changes)
        yield self._to_proto(user)

class AuthInterceptor(grpc.ServerInterceptor):
    def __init__(self, auth_service):
        self.auth_service = auth_service
        self.public_methods = {"/user.v1.UserService/CreateUser"}

    def intercept_service(self, continuation, handler_call_details):
        if handler_call_details.method not in self.public_methods:
            metadata = dict(handler_call_details.invocation_metadata)
            token = metadata.get("authorization", "").replace("Bearer ", "")
            if not token or not self.auth_service.verify(token):
                return grpc.unary_unary_rpc_method_handler(
                    lambda req, ctx: ctx.abort(grpc.StatusCode.UNAUTHENTICATED, "Invalid token")
                )
        return continuation(handler_call_details)

class RetryInterceptor(grpc.UnaryUnaryClientInterceptor):
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.retry_codes = {grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED}

    def intercept_unary_unary(self, continuation, client_call_details, request):
        for attempt in range(self.max_retries):
            try:
                return continuation(client_call_details, request)
            except grpc.RpcError as e:
                if e.code() not in self.retry_codes or attempt == self.max_retries - 1:
                    raise
                time.sleep(2 ** attempt)  # Exponential backoff

// Send message via WAHA (WhatsApp HTTP API)
const response = await fetch(`${WAHA_URL}/api/sendText`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    chatId: `${phone}@c.us`,
    text: message,
    session: "default",
  }),
});

// Set webhook for Telegram bot
await fetch(`https://api.telegram.org/bot${TOKEN}/setWebhook`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    url: `${BASE_URL}/api/telegram/webhook`,
    secret_token: WEBHOOK_SECRET,
  }),
});

// Telegram: verify secret_token header
function verifyTelegramWebhook(req: Request): boolean {
  return req.headers["x-telegram-bot-api-secret-token"] === WEBHOOK_SECRET;
}

// Payload 3.0 collection config
import type { CollectionConfig } from "payload";

export const Posts: CollectionConfig = {
  slug: "posts",
  admin: { useAsTitle: "title" },
  access: {
    read: () => true,
    create: ({ req: { user } }) => Boolean(user),
    update: ({ req: { user } }) => user?.role === "admin",
    delete: ({ req: { user } }) => user?.role === "admin",
  },
  fields: [
    { name: "title", type: "text", required: true },
    { name: "content", type: "richText" },
    { name: "author", type: "relationship", relationTo: "users" },
    { name: "status", type: "select", options: ["draft", "published"] },
  ],
  hooks: {
    beforeChange: [({ data }) => ({ ...data, updatedAt: new Date() })],
  },
};

# No keepalive, no abort handling, no reconnection support
@app.get("/stream")
async def stream():
    async def generate():
        for item in data:
            yield f"data: {item}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

export async function GET(req: Request) {
  const encoder = new TextEncoder()

  const stream = new ReadableStream({
    async start(controller) {
      controller.enqueue(encoder.encode('data: Hello\n\n'))

      // Keep connection alive every 30s
      const interval = setInterval(() => {
        controller.enqueue(encoder.encode(': keepalive\n\n'))
      }, 30000)

      // Cleanup on client disconnect
      req.signal.addEventListener('abort', () => {
        clearInterval(interval)
        controller.close()
      })
    }
  })

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    }
  })
}

export async function POST(req: Request) {
  const { messages } = await req.json()
  const stream = await openai.chat.completions.create({ model: 'gpt-5.2', messages, stream: true })
  const encoder = new TextEncoder()

  return new Response(
    new ReadableStream({
      async start(controller) {
        for await (const chunk of stream) {
          const content = chunk.choices[0]?.delta?.content
          if (content) {
            controller.enqueue(encoder.encode("data: " + JSON.stringify({ content }) + "\n\n"))
          }
        }
        controller.enqueue(encoder.encode('data: [DONE]\n\n'))
        controller.close()
      }
    }),
    { headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' } }
  )
}

class ReconnectingEventSource {
  private eventSource: EventSource | null = null
  private reconnectDelay = 1000
  private maxReconnectDelay = 30000

  constructor(private url: string, private onMessage: (data: string) => void) {
    this.connect()
  }

  private connect() {
    this.eventSource = new EventSource(this.url)
    this.eventSource.onmessage = (event) => {
      this.reconnectDelay = 1000
      this.onMessage(event.data)
    }
    this.eventSource.onerror = () => {
      this.eventSource?.close()
      setTimeout(() => this.connect(), this.reconnectDelay)
      this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxReconnectDelay)
    }
  }
}

# No heartbeat, no validation, no reconnection
wss.on('connection', (ws) => {
  ws.on('message', (data) => {
    wss.clients.forEach((client) => client.send(data))  // Raw broadcast!
  })
})

const wss = new WebSocketServer({ port: 8080 })

wss.on('connection', (ws) => {
  // Heartbeat
  const heartbeat = setInterval(() => ws.ping(), 30000)

  ws.on('message', (data) => {
    const parsed = JSON.parse(data.toString())
    // Validate message structure
    if (!parsed.type || !parsed.text) return

    // Broadcast to connected clients
    wss.clients.forEach((client) => {
      if (client !== ws && client.readyState === WebSocket.OPEN) {
        client.send(JSON.stringify(parsed))
      }
    })
  })

  ws.on('close', () => clearInterval(heartbeat))
})

# Generator not closed if exception occurs mid-iteration
async for chunk in external_api_stream():  # Resource leak if exception!
    yield process(chunk)

from contextlib import aclosing

# Guaranteed cleanup with aclosing()
async def stream_llm_response(prompt: str):
    async with aclosing(llm.astream(prompt)) as stream:
        async for chunk in stream:
            yield chunk.content

# Consumption with proper cleanup
async def consume():
    async with aclosing(stream_llm_response("Hello")) as response:
        async for token in response:
            handle(token)

from fastapi import Response
from datetime import date

def add_deprecation_headers(
    response: Response,
    deprecated_date: date,
    sunset_date: date,
    link: str,
):
    response.headers["Deprecation"] = deprecated_date.isoformat()
    response.headers["Sunset"] = sunset_date.isoformat()
    response.headers["Link"] = f'<{link}>; rel="successor-version"'

# Usage in v1 endpoints
@router.get("/users/{user_id}")
async def get_user_v1(user_id: str, response: Response):
    add_deprecation_headers(
        response,
        deprecated_date=date(2025, 1, 1),
        sunset_date=date(2025, 7, 1),
        link="https://api.example.com/v2/users",
    )
    return await service.get_user(user_id)

┌─────────────────────────────────────────────────────────────────┐
│                     VERSION LIFECYCLE                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  ┌─────────┐   ┌─────────┐   ┌──────────┐   ┌─────────────┐   │
│  │  ALPHA  │ → │  BETA   │ → │  STABLE  │ → │ DEPRECATED  │   │
│  │ (dev)   │   │ (test)  │   │ (prod)   │   │ (sunset)    │   │
│  └─────────┘   └─────────┘   └──────────┘   └─────────────┘   │
│                                                                 │
│  POLICY:                                                        │
│  • Deprecation notice: 3 months before sunset                   │
│  • Sunset period: 6 months after deprecation                    │
│  • Support: Latest stable + 1 previous version                  │
└─────────────────────────────────────────────────────────────────┘

# Adding optional fields
class UserResponse(BaseModel):
    id: str
    name: str
    avatar_url: str | None = None  # New optional field

# Adding new endpoints
@router.get("/users/{user_id}/preferences")  # New endpoint

# Adding optional query params
@router.get("/users")
async def list_users(
    limit: int = 100,
    cursor: str | None = None,  # New pagination
):

# Removing fields
# Renaming fields
# Changing field types
# Changing URL structure
# Changing authentication
# Removing endpoints
# Changing error formats

from starlette.middleware.base import BaseHTTPMiddleware

DEPRECATED_VERSIONS = {
    "v1": {
        "sunset": datetime(2025, 12, 31),
        "successor": "v2",
    }
}

class DeprecationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        response = await call_next(request)

        path = request.url.path
        for version, info in DEPRECATED_VERSIONS.items():
            if f"/api/{version}/" in path:
                response.headers["Deprecation"] = "true"
                response.headers["Sunset"] = info["sunset"].strftime(
                    "%a, %d %b %Y %H:%M:%S GMT"
                )
                successor_path = path.replace(
                    f"/api/{version}/",
                    f"/api/{info['successor']}/"
                )
                response.headers["Link"] = (
                    f'<{successor_path}>; rel="successor-version"'
                )
                break
        return response

app.add_middleware(DeprecationMiddleware)

# NEVER version internal implementation
class UserServiceV1:  # Services should be version-agnostic
    ...

# NEVER break contracts without versioning
class UserResponse(BaseModel):
    full_name: str  # Changed from `name` without version bump!

# NEVER sunset without notice
# Just removing v1 routes one day

# NEVER support too many versions (max 2-3)

# v1 schema changed without versioning
class UserResponse(BaseModel):
    id: str
    full_name: str  # Changed from "name" - BREAKS clients!

# v1 stays unchanged
class UserResponseV1(BaseModel):
    id: str
    name: str

# v2 with breaking changes
class UserResponseV2(BaseModel):
    id: str
    first_name: str
    last_name: str

from fastapi import Header, HTTPException, Depends

SUPPORTED_VERSIONS = {1, 2}
DEFAULT_VERSION = 2

async def get_api_version(
    x_api_version: str = Header(default="1", alias="X-API-Version")
) -> int:
    try:
        version = int(x_api_version)
        if version not in SUPPORTED_VERSIONS:
            raise ValueError()
        return version
    except ValueError:
        raise HTTPException(
            400,
            f"Invalid API version. Supported: {SUPPORTED_VERSIONS}",
        )

@router.get("/users/{user_id}")
async def get_user(
    user_id: str,
    version: int = Depends(get_api_version),
    service: UserService = Depends(),
):
    user = await service.get_user(user_id)

    if version == 1:
        return UserResponseV1(id=user.id, name=user.full_name)
    else:
        return UserResponseV2(
            id=user.id,
            first_name=user.first_name,
            last_name=user.last_name,
        )

from fastapi import Request

MEDIA_TYPES = {
    "application/vnd.orchestkit.v1+json": 1,
    "application/vnd.orchestkit.v2+json": 2,
    "application/json": 2,  # Default to latest
}

async def get_version_from_accept(request: Request) -> int:
    accept = request.headers.get("Accept", "application/json")
    return MEDIA_TYPES.get(accept, 2)

@router.get("/users/{user_id}")
async def get_user(
    user_id: str,
    version: int = Depends(get_version_from_accept),
):
    ...

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_header_versioning(client: AsyncClient):
    # Request with v1 header
    response = await client.get(
        "/api/users/123",
        headers={"X-API-Version": "1"},
    )
    data = response.json()
    assert "avatar_url" not in data

    # Request with v2 header
    response = await client.get(
        "/api/users/123",
        headers={"X-API-Version": "2"},
    )
    data = response.json()
    assert "avatar_url" in data

# Breaks when header missing
async def get_version(x_api_version: str = Header()):
    return int(x_api_version)  # Error if header absent!

# Falls back to latest stable version
async def get_version(
    x_api_version: str = Header(default="2")
) -> int:
    return int(x_api_version)

backend/app/
├── api/
│   ├── v1/
│   │   ├── __init__.py
│   │   ├── routes/
│   │   │   ├── users.py
│   │   │   └── analyses.py
│   │   └── schemas/
│   │       ├── user.py
│   │       └── analysis.py
│   ├── v2/
│   │   ├── __init__.py
│   │   ├── routes/
│   │   │   ├── users.py      # Updated schemas
│   │   │   └── analyses.py
│   │   └── schemas/
│   │       ├── user.py       # New schema version
│   │       └── analysis.py
│   └── router.py             # Combines all versions
├── core/
└── services/                  # Shared across versions

# backend/app/api/router.py
from fastapi import APIRouter
from app.api.v1.router import router as v1_router
from app.api.v2.router import router as v2_router

api_router = APIRouter()
api_router.include_router(v1_router, prefix="/v1")
api_router.include_router(v2_router, prefix="/v2")

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

# v1/schemas/user.py
class UserResponseV1(BaseModel):
    id: str
    name: str  # Single name field

# v2/schemas/user.py
class UserResponseV2(BaseModel):
    id: str
    first_name: str  # Split into first/last
    last_name: str
    full_name: str   # Computed for convenience

# services/user_service.py (version-agnostic)
class UserService:
    async def get_user(self, user_id: str) -> User:
        return await self.repo.get_by_id(user_id)

# v1/routes/users.py
@router.get("/{user_id}", response_model=UserResponseV1)
async def get_user_v1(user_id: str, service: UserService = Depends()):
    user = await service.get_user(user_id)
    return UserResponseV1(id=user.id, name=user.full_name)

# v2/routes/users.py
@router.get("/{user_id}", response_model=UserResponseV2)
async def get_user_v2(user_id: str, service: UserService = Depends()):
    user = await service.get_user(user_id)
    return UserResponseV2(
        id=user.id,
        first_name=user.first_name,
        last_name=user.last_name,
        full_name=f"{user.first_name} {user.last_name}",
    )

# Services should be version-agnostic
class UserServiceV1:
    async def get_user(self, id: str):
        ...

class UserServiceV2:
    async def get_user(self, id: str):
        ...