DOCSEARCHbyAlgolia
Submit Docs

svc-infra

repository·main·Indexed 41 days ago

https://github.com/nfraxlab/svc-infra

Infrastructure for building and deploying production-ready FastAPI services. Version 1.22.2 provides components for authentication, billing, background jobs, webhooks, observability, and storage. It includes features like auto-CRUD REST endpoints via SqlResource, branded auth email templates, and support for multiple email backends (Resend, SendGrid, SES, Postmark, SMTP).

Tokens
230.5K
Snippets
659
Records
838
Agent score
98%

What's inside svc-infra

  1. Overview of Generic Document Management

    main

    The svc_infra.documents module provides a domain-agnostic system for storing and managing document metadata. It is designed to work with any storage backend supported by svc-infra (S3, local, or memory) and includes built-in user scoping for multi-tenant applications.

    Key Features:

    • Generic Document Model: Flexible metadata schema.
    • Storage Integration: Seamlessly uses S3, local, or memory backends.
    • FastAPI Endpoints: Ready-to-use protected routes for CRUD operations.
    • Async-First: Full async/await support.
    • User Isolation: Built-in scoping to ensure users only access their own documents.

    Note: This module does not include OCR, AI analysis, file format conversion, or virus scanning. These should be implemented as extensions.

  2. Overview of svc-infra features

    main

    svc-infra provides a comprehensive suite of backend primitives for building production-ready Python services:

    • API Framework: FastAPI-based scaffolding with dual routers, automatic OpenAPI generation, and a middleware stack.
    • Authentication: Support for OAuth2, sessions, MFA, password policies, JWT tokens, and user management.
    • Database: SQLAlchemy ORM, Alembic migrations, connection pooling, and multi-database support.
    • Caching: Redis caching, cache decorators, TTL management, and invalidation strategies.
    • Background Jobs: Job queues, workers, schedulers, and retry logic for async processing.
    • Webhooks: Webhook signing, delivery, retry, and subscription management.
    • Deployment: Platform detection for Railway, Render, AWS, GCP, and Azure with environment resolution.
    • Logging: Structured JSON logging, context injection, and container-optimized output.
    • Testing: Mock cache, mock job queue, and fixture factories for testing without external services.
    • Developer Experience: CI workflow generation, changelog tools, and code quality checks.

  3. Overview of svc-infra CLI command groups

    main

    The svc-infra CLI is organized into several functional groups:

    GroupPurpose
    sqlSQL database migrations (Alembic), scaffolding, export
    mongoMongoDB operations and scaffolding
    dbDatabase operations (wait, kill-queries)
    obsObservability stack (Grafana, Prometheus)
    jobsBackground job scheduler
    sdkClient SDK generation (TypeScript, Python, Postman)
    dxDeveloper experience (CI, OpenAPI checks, changelog)
    healthHealth endpoint verification
    docsInteractive documentation browser

  4. Overview of the svc_infra.storage system

    main

    The svc_infra.storage module provides a backend-agnostic abstraction for file storage. It allows you to write code that interacts with a unified interface, enabling you to switch between different storage providers (Local, S3, GCS, Cloudinary, or In-Memory) without changing your application logic.

    Key features include:

    • Backend Abstraction: Use a single API for multiple providers.
    • Signed URLs: Generate secure, time-limited access to files.
    • Metadata Support: Attach custom metadata (e.g., user_id, tags) to files.
    • FastAPI Integration: Easy setup via dependency injection.
    • Auto-detection: Automatically configures the backend based on environment variables.

  5. The StorageBackend protocol interface

    main

    All storage backends implement the StorageBackend protocol. When using the storage system, you will interact with these core methods:

    • put(key, data, content_type, metadata): Stores a file and returns its URL.
    • get(key): Retrieves the raw file content.
    • delete(key): Removes the file associated with the key.
    • exists(key): Checks if a file exists at the given key.
    • get_url(key, expires_in, download): Generates a signed or public URL.
    • list_keys(prefix, limit): Lists stored files matching a prefix.
    • get_metadata(key): Retrieves metadata attached to the file.

  6. Ensure idempotency for mutating requests

    main

    To prevent duplicate operations (e.g., double charging a customer), all mutating endpoints (POST, PUT, DELETE) require an Idempotency-Key header.

    This is implemented via IdempotencyMiddleware. While it uses an in-memory store by default, you should plug in Redis for production environments to ensure consistency across multiple service instances. Use a unique key per logical operation, such as f"order-{order_id}-payment".

    headers = {"Idempotency-Key": "unique-key-per-request"}

  7. Planned AI Integration: Semantic Search

    main

    Documents can be indexed for semantic search using the ai-infra Retriever. The workflow involves:

    1. Iterating through documents via DocumentManager.
    2. Extracting text (via OCR if necessary).
    3. Adding text to the Retriever with metadata (e.g., document_id, user_id, filename).
    4. Searching using retriever.search() with filters (e.g., user_id) and enriching results with metadata from the DocumentManager.
    from ai_infra import Retriever
from svc_infra.documents import easy_documents

# Indexing pattern
async def index_documents_for_search(user_id: str, retriever: Retriever):
    manager = easy_documents()
    docs = manager.list(user_id)
    for doc in docs:
        text = await extract_text_from_document(doc.id)
        retriever.add_text(
            text,
            metadata={
                "source": f"document://{doc.id}",
                "document_id": doc.id,
                "user_id": doc.user_id,
                **doc.metadata
            }
        )

# Searching pattern
async def search_documents(user_id: str, query: str, retriever: Retriever, limit: int = 10):
    results = retriever.search(query, filter={"user_id": user_id}, limit=limit)
    # ... enrichment logic

  8. Define Database Models with ModelBase

    main

    To ensure your models are discovered by the svc-infra migration system, they must inherit from svc_infra.db.sql.base.ModelBase. Inheriting from standard SQLAlchemy DeclarativeBase will prevent migrations from detecting your tables.

    Correct Pattern

    from svc_infra.db.sql.base import ModelBase

class Project(ModelBase):
    __tablename__ = "projects"
    # ...
    from svc_infra.db.sql.base import ModelBase

class Project(ModelBase):
    __tablename__ = "projects"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)

  9. Connect MCP OAuth servers via automatic discovery

    main

    svc_infra.connect supports automatic OAuth discovery for Model Context Protocol (MCP) servers that implement the OAuth Authorization Framework (RFC 9728 and RFC 8414).

    When a client provides an mcp_server_url to the /connect/authorize endpoint, the system performs the following steps:

    1. Probes https://<mcp_server_url>/.well-known/oauth-protected-resource to find authorization_servers.
    2. Fetches metadata from the first authorization server via /.well-known/oauth-authorization-server to obtain authorization_endpoint and token_endpoint.
    3. Registers the discovered OAuthProvider in the global registry under the key mcp:<url>.
    4. Executes a standard PKCE flow.

    Discovered providers are cached in-process for 1 hour.

    GET /connect/authorize?mcp_server_url=https://mcp.example.com&connection_id=<id>

  10. When to use extract_router() vs manual definition

    main

    Use extract_router() when:

    • Building a monolithic versioned API where all features belong under a prefix like /v0 or /v1.
    • You want unified documentation at /v0/docs showing all features together.
    • You are consolidating multiple integrations under one version.
    • You need version-specific behavior for third-party integrations.

    Do NOT use extract_router() when:

    • The feature should have its own root-level endpoint (e.g., public webhooks at /webhooks).
    • The integration is shared across multiple versions (mount it at the root instead).
    • You only need a small subset of the integration's endpoints.

    Use Manual Definition when:

    If you only need a subset of endpoints or want custom validation/transforms per endpoint, define them manually using a public_router or similar:

    from svc_infra.api.fastapi.dual.public import public_router
from fin_infra.banking import easy_banking

router = public_router(prefix="/banking", tags=["Banking"])
banking = easy_banking(provider="plaid")

@router.post("/link")
async def create_link(request: CreateLinkRequest):
    return banking.create_link_token(user_id=request.user_id)

  11. How APF Payments providers and auth postures work

    main

    Providers

    Providers are adapters implementing the ProviderAdapter protocol. The active provider is selected via the PAYMENTS_PROVIDER environment variable. Built-in providers include Stripe (stripe) and Aiydan (aiydan).

    Router Auth Postures

    Endpoints are grouped by the type of authentication required:

    • Public: No auth required (e.g., /payments/webhooks/{provider}).
    • User: Requires a User JWT (e.g., /payments/customers, /payments/intents).
    • Protected: Requires either User or Service auth (e.g., /payments/methods).
    • Service: Requires Service-to-Service auth via API key (e.g., /payments/products, /payments/prices).

  12. Implement a Circuit Breaker to prevent cascading failures

    main

    The CircuitBreaker pattern prevents repeated calls to a failing service by

    from svc_infra.resilience import CircuitBreaker

breaker = CircuitBreaker(
    name="payment-api",
    failure_threshold=5,      # Open after 5 failures
    recovery_timeout=30.0,    # Wait 30s before trying again
)

async with breaker:
    result = await payment_api.charge(amount)