Merge pull request #25 from lsst-sqre/tickets/DM-39646

DM-39646: Add FastAPI app integration through dependencies

.github/workflows/ci.yaml

@@ -1,6 +1,7 @@
 name: Python CI
 
 "on":
+  merge_group: {}
   push:
     branches-ignore:
       # These should always correspond to pull requests, so ignore them for
@@ -11,9 +12,9 @@ name: Python CI
       - "renovate/**"
       - "tickets/**"
       - "u/**"
-    tags:
-      - "*"
   pull_request: {}
+  release:
+    types: [published]
 
 jobs:
   lint:
@@ -37,8 +38,6 @@ jobs:
     strategy:
       matrix:
         python:
-          - "3.8"
-          - "3.9"
           - "3.10"
           - "3.11"
 
@@ -81,22 +80,48 @@ jobs:
           username: ${{ secrets.LTD_USERNAME }}
           password: ${{ secrets.LTD_PASSWORD }}
         if: >
-          github.event_name != 'pull_request'
-          || startsWith(github.head_ref, 'tickets/')
+          github.event_name != 'merge_group'
+          && (github.event_name != 'pull_request'
+              || startsWith(github.head_ref, 'tickets/'))
+
+  test-packaging:
+
+    name: Test packaging
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          fetch-depth: 0 # full history for setuptools_scm
+
+      - name: Build and publish
+        uses: lsst-sqre/build-and-publish-to-pypi@v2
+        with:
+          python-version: "3.11"
+          upload: false
 
   pypi:
 
+    # This job requires set up:
+    # 1. Set up a trusted publisher for PyPI
+    # 2. Set up a "pypi" environment in the repository
+    # See https://github.com/lsst-sqre/build-and-publish-to-pypi
+    name: Upload release to PyPI
     runs-on: ubuntu-latest
-    needs: [lint, test, docs]
+    needs: [lint, test, docs, test-packaging]
+    environment:
+      name: pypi
+      url: https://pypi.org/p/kafkit
+    permissions:
+      id-token: write
+    if: github.event_name == 'release' && github.event.action == 'published'
 
     steps:
       - uses: actions/checkout@v3
         with:
           fetch-depth: 0 # full history for setuptools_scm
 
       - name: Build and publish
-        uses: lsst-sqre/build-and-publish-to-pypi@v1
+        uses: lsst-sqre/build-and-publish-to-pypi@v2
         with:
-          pypi-token: ${{ secrets.PYPI_SQRE_ADMIN }}
           python-version: "3.11"
-          upload: ${{ github.event_name == 'push' && startsWith(github.ref, 'refs/tags/') }}

.github/workflows/dependencies.yaml

@@ -0,0 +1,33 @@
+name: Dependency Update
+
+"on":
+  schedule:
+    - cron: "0 12 * * 1"
+  workflow_dispatch: {}
+
+jobs:
+  update:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Run neophile
+        uses: lsst-sqre/run-neophile@v1
+        with:
+          python-version: "3.11"
+          mode: pr
+          types: pre-commit
+          app-id: ${{ secrets.NEOPHILE_APP_ID }}
+          app-secret: ${{ secrets.NEOPHILE_PRIVATE_KEY }}
+
+      - name: Report status
+        if: always()
+        uses: ravsamhq/notify-slack-action@v2
+        with:
+          status: ${{ job.status }}
+          notify_when: "failure"
+          notification_title: "Periodic dependency update for {repo} failed"
+        env:
+          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_ALERT_WEBHOOK }}

.github/workflows/periodic-ci.yaml

@@ -15,8 +15,6 @@ jobs:
     strategy:
       matrix:
         python:
-          - "3.8"
-          - "3.9"
           - "3.10"
           - "3.11"
 
@@ -43,17 +41,17 @@ jobs:
           tox-envs: "docs,docs-linkcheck"
           use-cache: false
 
-  pypi:
+  test-packaging:
     runs-on: ubuntu-latest
+    timeout-minutes: 10
 
     steps:
       - uses: actions/checkout@v3
         with:
           fetch-depth: 0 # full history for setuptools_scm
 
       - name: Build and publish
-        uses: lsst-sqre/build-and-publish-to-pypi@v1
+        uses: lsst-sqre/build-and-publish-to-pypi@v2
         with:
-          pypi-token: ""
           python-version: "3.11"
           upload: false

.gitignore

@@ -1,3 +1,5 @@
+.vscode
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

Makefile

@@ -5,7 +5,7 @@ help:
 
 .PHONY: init
 init:
-	pip install -e ".[aiohttp,httpx,pydantic,dev]"
-	pip install -U tox pre-commit
+	pip install -e ".[aiohttp,httpx,pydantic,aiokafka,dev]"
+	pip install -U tox pre-commit scriv
 	pre-commit install
 	rm -rf .tox

changelog.d/20230717_125707_jsick_DM_39646.md

@@ -0,0 +1,16 @@
+### Backwards-incompatible changes
+
+- Only Python 3.10 or later is supported.
+
+### New features
+
+- Integration into FastAPI apps through dependencies in `kafkit.fastapi.dependencies`:
+
+  - `AioKafkaProducerDependency` provides a Kafka producer based on aiokafka's `AIOKafkaProducer` (requires the `aiokafka` extra).
+  - `PydanticSchemaManager` provides a Pydantic-based schema manager for Avro schemas, `kafkit.schema.manager.PydanticSchemaManager`.
+  - `RegistryApiDependency` provides an HTTPX-based Schema Registry client, `kafkit.registry.httpx.RegistryApi`.
+
+### Other changes
+
+- Adopt PyPI's trusted publishers mechanism for releases.
+- Adopt the new [Neophile](https://github.com/lsst-sqre/neophile) workflow for keeping pre-commit hooks up-to-date.

docs/api.rst

@@ -2,6 +2,18 @@
 Kafkit API reference
 ####################
 
+.. automodapi:: kafkit.fastapi.dependencies.aiokafkaproducer
+   :no-inheritance-diagram:
+   :include-all-objects:
+
+.. automodapi:: kafkit.fastapi.dependencies.pydanticschemamanager
+   :no-inheritance-diagram:
+   :include-all-objects:
+
+.. automodapi:: kafkit.fastapi.dependencies.registryapi
+   :no-inheritance-diagram:
+   :include-all-objects:
+
 .. automodapi:: kafkit.registry
    :no-inheritance-diagram:
 

pyproject.toml

@@ -14,23 +14,22 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Intended Audience :: Developers",
     "Natural Language :: English",
     "Operating System :: POSIX",
     "Typing :: Typed",
 ]
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 dependencies = ["fastavro", "uritemplate"]
 dynamic = ["version"]
 
 [project.optional-dependencies]
 aiohttp = ["aiohttp"]
 httpx = ["httpx"]
 pydantic = ["pydantic", "dataclasses-avroschema[pydantic]"]
+aiokafka = ["aiokafka"]
 dev = [
     # Testing
     "coverage[toml]",
@@ -82,7 +81,7 @@ exclude_lines = [
 
 [tool.black]
 line-length = 79
-target-version = ['py38']
+target-version = ['py310']
 exclude = '''
 /(
     \.eggs

src/kafkit/fastapi/__init__.py

@@ -0,0 +1 @@
+"""Kafkit integration with FastApi applications."""

src/kafkit/fastapi/dependencies/__init__.py

@@ -0,0 +1 @@
+"""FastAPI dependencies for Kafkit applications."""

src/kafkit/fastapi/dependencies/aiokafkaproducer.py

@@ -0,0 +1,56 @@
+"""A FastAPI dependency that provides an aiokafka Producer."""
+
+import aiokafka  # patched for testing
+
+from kafkit.settings import KafkaConnectionSettings
+
+__all__ = ["kafka_producer_dependency", "AioKafkaProducerDependency"]
+
+
+class AioKafkaProducerDependency:
+    """A FastAPI dependency that provides an aiokafka Producer."""
+
+    def __init__(self) -> None:
+        self._producer: aiokafka.AIOKafkaProducer | None = None
+
+    async def initialize(self, settings: KafkaConnectionSettings) -> None:
+        """Initialize the dependency (call during FastAPI startup).
+
+        Parameters
+        ----------
+        settings
+            The Kafka connection settings.
+        """
+        security_protocol = settings.security_protocol.value
+        sasl_mechanism = (
+            settings.sasl_mechanism.value if settings.sasl_mechanism else None
+        )
+        self._producer = aiokafka.AIOKafkaProducer(
+            bootstrap_servers=settings.bootstrap_servers,
+            security_protocol=security_protocol,
+            ssl_context=settings.ssl_context,
+            sasl_mechanism=sasl_mechanism,
+            sasl_plain_password=(
+                settings.sasl_password.get_secret_value()
+                if settings.sasl_password
+                else None
+            ),
+            sasl_plain_username=settings.sasl_username,
+        )
+        await self._producer.start()
+
+    async def __call__(self) -> aiokafka.AIOKafkaProducer:
+        """Get the dependency (call during FastAPI request handling)."""
+        if self._producer is None:
+            raise RuntimeError("Dependency not initialized")
+        return self._producer
+
+    async def stop(self) -> None:
+        """Stop the dependency (call during FastAPI shutdown)."""
+        if self._producer is None:
+            raise RuntimeError("Dependency not initialized")
+        await self._producer.stop()
+
+
+kafka_producer_dependency = AioKafkaProducerDependency()
+"""The FastAPI dependency callable that provides an AIOKafkaProducer."""

src/kafkit/fastapi/dependencies/pydanticschemamanager.py

@@ -0,0 +1,72 @@
+"""A FastAPI dependency that provides a Kafkit PydanticSchemaManager
+for serializing Pydantic models into Avro.
+"""
+
+from collections.abc import Iterable
+from typing import Type
+
+from dataclasses_avroschema.avrodantic import AvroBaseModel
+from httpx import AsyncClient
+
+from kafkit.registry import manager  # this is patched in tests
+from kafkit.registry.httpx import RegistryApi
+
+__all__ = [
+    "pydantic_schema_manager_dependency",
+    "PydanticSchemaManagerDependency",
+]
+
+
+class PydanticSchemaManagerDependency:
+    """A FastAPI dependency that provides a Kafkit PydanticSchemaManager
+    for serializing Pydantic models into Avro.
+    """
+
+    def __init__(self) -> None:
+        self._schema_manager: manager.PydanticSchemaManager | None = None
+
+    async def initialize(
+        self,
+        *,
+        http_client: AsyncClient,
+        registry_url: str,
+        models: Iterable[Type[AvroBaseModel]],
+        suffix: str = "",
+        compatibility: str = "FORWARD",
+    ) -> None:
+        """Initialize the dependency (call during FastAPI startup).
+
+        Parameters
+        ----------
+        http_client
+            The httpx AsyncClient instance to use for HTTP requests.
+        registry_url
+            The URL of the Schema Registry.
+        models
+            The Pydantic models to register.
+        suffix
+            A suffix that is added to the schema name (and thus subject name),
+            for example ``_dev1``.
+        compatibility
+            The compatibility level to use when registering the schemas.
+        """
+        registry_api = RegistryApi(http_client=http_client, url=registry_url)
+        self._schema_manager = manager.PydanticSchemaManager(
+            registry=registry_api, suffix=suffix
+        )
+
+        await self._schema_manager.register_models(
+            models, compatibility=compatibility
+        )
+
+    async def __call__(self) -> manager.PydanticSchemaManager:
+        """Get the dependency (call during FastAPI request handling)."""
+        if self._schema_manager is None:
+            raise RuntimeError("Dependency not initialized")
+        return self._schema_manager
+
+
+pydantic_schema_manager_dependency = PydanticSchemaManagerDependency()
+"""The FastAPI dependency callable that provides a Kafkit PydanticSchemaManager
+instance for serializing Pydantic models into Avro.
+"""