flask-integration

A small framework for building integrations in Flask applications:

• key-authenticated JSON API endpoints, declared with decorators and typed with Pydantic v2
• outgoing webhooks with typed payloads — subscribed URLs are called when your app triggers the event
• an OpenAPI 3.1 document per integration (including the native webhooks section) at <base>/openapi.json
• a flask integration CLI to manage API keys and webhook subscriptions and to inspect what's declared in code
• storage in your app's database, with schema contributed to live-alembic via the live_alembic.models entry point
• an optional debug log of every action the framework takes

Requires Python ≥ 3.12, Flask ≥ 3, Pydantic ≥ 2, SQLAlchemy ≥ 2.

Declaring an integration

from flask import Flask
from pydantic import BaseModel
from flask_integration import FlaskIntegration, FlaskIntegrationBase, api_endpoint, webhook


class CreateOrder(BaseModel):
    sku: str
    quantity: int

class Order(BaseModel):
    order_id: int
    sku: str
    quantity: int

class OrderCreated(BaseModel):
    order_id: int
    total: float


class ShopIntegration(FlaskIntegrationBase):
    name = "shop"                      # default: class name without "Integration", snake_case
    title = "Shop integration API"     # OpenAPI title

    @api_endpoint(methods=["POST"], summary="Create an order")
    def create_order(self, body: CreateOrder) -> Order:
        # request body validated into CreateOrder; returned model is serialized
        return Order(order_id=1, sku=body.sku, quantity=body.quantity)

    @api_endpoint                       # GET /ping
    def ping(self):
        """Liveness check."""
        return {"pong": True}

    @webhook(payload=OrderCreated)
    def order_created(self):
        """Fired when an order is created."""


app = Flask(__name__)
app.config["INTEGRATION_DATABASE_URI"] = "sqlite:///app.db"

shop = ShopIntegration()
FlaskIntegration(app, shop, url_prefix="/api/integrations/shop")

• @api_endpoint infers the request body model from the first Pydantic-annotated parameter and the response model from the return annotation (request_model= / response_model= override). The URL rule defaults to the method name with hyphens; pass a rule explicitly with @api_endpoint("/orders", methods=["POST"]).
• Parameters named in the rule receive the URL values, with Flask converters: @api_endpoint("/spaces/<space>/pages/<int:page_id>") → def page(self, space: str, page_id: int). OpenAPI shows /spaces/{space}/pages/{page_id}.
• A model parameter is read from the query string when it is marked Annotated[Model, Query]; on an endpoint without POST / PUT / PATCH the first model parameter is too. Repeated keys and list fields become lists, booleans accept yes/true/1. query_model= overrides. A further model parameter on such an endpoint would be a request body and is a TypeError at registration (unless request_model= is given).
• The return annotation can be any type pydantic serializes — -> list[Order], -> dict[str, Order] — and is documented as such; -> None documents a 204.
• Raise ApiError(404, "not_found", "no such order") (or a subclass with status / code class attributes, raised as NotFound(detail=...)) to answer {"error": "not_found", "detail": "no such order"}. abort() and the 401 / 422 responses use the same shape. Document them with @api_endpoint(..., errors={404: "No such order"}).

from typing import Annotated
from flask_integration import ApiError, Query

class OrderFilter(BaseModel):
    status: list[str] = []
    limit: int = 50

class ShopIntegration(FlaskIntegrationBase):
    @api_endpoint("/orders")                     # GET /orders?status=open&limit=10
    def list_orders(self, filters: OrderFilter) -> list[Order]: ...

    @api_endpoint("/orders/<int:order_id>", errors={404: "No such order"})
    def get_order(self, order_id: int) -> Order:
        raise ApiError(404, "not_found", f"no order {order_id}")

    @api_endpoint("/orders/<int:order_id>", methods=["PATCH"])
    def update_order(self, order_id: int, body: CreateOrder,
                     opts: Annotated[OrderFilter, Query]) -> Order: ...

• Subclassing an integration inherits its endpoints and webhooks. Overriding an endpoint method without the decorator keeps the endpoint and runs the override; ping = None removes it. An explicit name is inherited, a derived one is derived again from the subclass's name.
• status_code=201 sets the success status; return (value, status), (value, headers) or (value, status, headers) to vary it per call while keeping the typed serialization. -> bytes is served as application/octet-stream, media_type="image/png" sets another type (both documented in OpenAPI), and a returned Response passes through.
• Call conditional(etag, last_modified=...) before the expensive part of a GET: a matching If-None-Match / If-Modified-Since ends the request with 304, otherwise the validators go on the response.
• @webhook(payload=Model) turns the method into the trigger. Anywhere in your app (inside an app context):

shop.order_created(OrderCreated(order_id=99, total=123.45))
# also accepts a dict or kwargs; payload is validated before sending
# returns the number of deliveries dispatched

Every URL subscribed to shop.order_created receives a POST with the JSON payload and X-Integration-Event, X-Integration-Delivery-Id (same on retries) and X-Integration-Timestamp headers. With a secret it is signed: X-Integration-Signature-V2: sha256=<hmac> over <delivery id>.<timestamp>.<body> (check with verify_signature(secret, headers, body), which rejects stale timestamps), plus a body-only X-Integration-Signature (no replay protection). Delivery runs on a background thread pool with retries (3 attempts, backoff), so triggering never blocks the caller; each delivery's outcome is kept in a delivery log (shop.list_deliveries(), flask integration list-deliveries).

Subscriptions can be scoped to a partition (--partition docs) and managed from Python (shop.subscribe(...), and flask_integration.subscriptions.update_subscription(app, id, ...) to rotate a secret or pause). Deliveries can run on the host's queue and HTTP client: FlaskIntegration(..., deliver=lambda job: queue.enqueue(run_delivery, job), send=my_post). The default sender caps each attempt at the delivery timeout in total, never reads the response body and does not follow redirects, so a hostile receiver cannot hold a worker; it will post to internal addresses, though, so let only operators subscribe unless send is a client with SSRF guards.

Pass session=db.session to trigger inside the caller's transaction: the push waits for the commit and is dropped on rollback, and the trigger returns None.

Event log and paging

@event(payload=Model, partition="field") is a webhook that is first written to a durable, ordered log, so consumers can resume from a cursor instead of relying on pushes:

class PublishIntegration(FlaskIntegrationBase):
    @event(payload=PageChanged, partition="space")
    def page_changed(self): ...

    @api_endpoint("/spaces/<space>/changes", errors={410: "Cursor too old"})
    def changes(self, space: str, q: Annotated[EventQuery, Query]) -> EventPage:
        return self.read_events(q.since, partition=space, limit=q.limit)

publish = PublishIntegration()
rec = publish.page_changed(space="docs", slug="home", session=db.session)
db.session.commit()   # row commits with the change; push (X-Integration-Event-Id: seq) follows

read_events returns items, next_since and more; a since older than retention (purge_events(app, timedelta(days=30)) or flask integration purge-events --older-than-days 30) gets 410 cursor_expired. latest_event_seq() is where a consumer starts after a full sync.

self.event_stream(since, partition=...) serves the log as Server-Sent Events (id: = seq, Last-Event-ID resumes, heartbeats, bounded lifetime), from an endpoint declared with media_type="text/event-stream". On the consumer side, flask_integration.receiver has iter_event_pages, stream_events, verify_signature and CursorExpired.

For listings, Paged[T] + PageQuery + paginate(session, stmt, Model.key, cursor, limit) (or paginate_items for lists) give keyset paging with opaque cursors.

Authentication

By default all endpoints require an API key in X-API-Key (or Authorization: Bearer …). Keys are stored as SHA-256 hashes; only a short prefix remains visible. The OpenAPI document is public by default (INTEGRATION_OPENAPI_REQUIRE_KEY = True to lock it down).

Whoever authenticated is the principal: mark a parameter Annotated[ApiKeyIdentity, Principal] to receive it, or call current_principal(). To use the host app's own tokens or users instead of the key table, pass an authenticator — a callable returning the principal, or None for a 401 (raise ApiError(403, …) to forbid):

FlaskIntegration(
    app, shop, url_prefix="/api/v1",
    authenticate=token_user,   # reads flask.request, returns e.g. a User
    security_schemes={"BearerAuth": {"type": "http", "scheme": "bearer"}},
)

Request bodies must be application/json (415 otherwise, 400 for invalid JSON). An authenticator that accepts a session cookie still needs the host's CSRF protection for endpoints without a body.

CLI

flask integration create-key ci-bot          # prints the key once
flask integration list-keys
flask integration revoke-key ci-bot

flask integration add-webhook shop order_created https://receiver/hook --secret s3cret
flask integration list-webhooks              # --json --show-secrets to include secrets
flask integration set-webhook 1 --secret n3w --inactive
flask integration remove-webhook 1
flask integration list-deliveries --limit 20
flask integration purge-deliveries --older-than-days 14

flask integration purge-events --older-than-days 30

flask integration list-endpoints             # everything declared in code

add-webhook validates the integration and event names against what is actually declared in code.

Database

The framework needs INTEGRATION_DATABASE_URI (falls back to SQLALCHEMY_DATABASE_URI) — or the host's own engine, FlaskIntegration(app, shop, engine=db.engine) (a callable such as lambda: db.engine is resolved on first use) — and owns five tables: flask_integration_api_keys, flask_integration_webhooks, the delivery log flask_integration_deliveries, and for the event log flask_integration_events and flask_integration_event_marks. Timestamps are stored as naive UTC and returned timezone-aware.

With live-alembic installed in the host app, the tables are contributed automatically through the entry point declared in this package:

[project.entry-points."live_alembic.models"]
flask_integration = "flask_integration.models"

so a plain livealembic migrate --file migrate_env.py in the host app creates them. Without live-alembic, create them yourself:

from flask_integration.models import Base
from flask_integration.db import get_engine

Base.metadata.create_all(get_engine(app))

OpenAPI

Each integration serves <url_prefix>/openapi.json. openapi_route="/spec.json" moves it; openapi_route=None turns it off for a host that serves its own document, which can then merge the integration in:

from flask_integration import merge_openapi

publish_api = FlaskIntegration(app, PublishIntegration(), url_prefix="/api/v1", openapi_route=None)
merge_openapi(host_spec, publish_api.openapi_spec(server_url="/api/v1"))

server_url (a path prefix of url_prefix) makes the paths relative to the host's server; conflicting paths, schemas or schemes raise ValueError.

Debug action log

app.config["INTEGRATION_DEBUG_LOG"] = True            # log to stderr
app.config["INTEGRATION_DEBUG_LOG_FILE"] = "fi.log"   # ... or to a file

Logs every action: key creation/revocation, auth successes and failures, endpoint calls, validation failures, webhook triggers, and each delivery attempt with its outcome (logger name: flask_integration.actions). It carries no keys, secrets, payloads or rejected input values, and webhook URLs only as scheme, host and port.

Config reference

Example app

See example/app.py for a runnable host app and example/migrate_env.py for the live-alembic setup.