Test FastStream 0.6.0

Author: dolfinus

.pre-commit-config.yaml

@@ -91,6 +91,7 @@ repos:
     rev: 0.8.17
     hooks:
       - id: uv-lock
+        args: [--prerelease=allow]
 
   - repo: local
     hooks:

data_rentgen/consumer/__init__.py

@@ -8,16 +8,18 @@
 import anyio
 from fast_depends import dependency_provider
 from faststream import ContextRepo, FastStream
-from faststream._compat import ExceptionGroup
+from faststream._internal._compat import ExceptionGroup
 from faststream.asgi import AsgiFastStream, AsgiResponse, get
 from faststream.kafka import KafkaBroker
-from faststream.kafka.publisher.asyncapi import AsyncAPIDefaultPublisher
+from faststream.kafka.publisher import DefaultPublisher
+from faststream.kafka.subscriber.usecase import BatchSubscriber
+from faststream.specification.asyncapi import AsyncAPI
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import data_rentgen
 from data_rentgen.consumer.settings import ConsumerApplicationSettings
 from data_rentgen.consumer.subscribers import runs_events_subscriber
-from data_rentgen.db.factory import create_session_factory
+from data_rentgen.db.factory import session_generator
 from data_rentgen.logging.setup_logging import setup_logging
 
 logger = logging.getLogger(__name__)
@@ -43,21 +45,34 @@ def broker_factory(settings: ConsumerApplicationSettings) -> KafkaBroker:
     # register subscribers using settings
     consumer_settings = settings.consumer.model_dump(exclude={"topics_list", "topics_pattern", "malformed_topic"})
 
-    subscriber = broker.subscriber(
+    subscribe = broker.subscriber(
         *settings.consumer.topics_list,
         pattern=settings.consumer.topics_pattern,
         **consumer_settings,
         batch=True,
         # Disable parsing JSONs on FastStream level
         decoder=lambda _: None,
     )
+
+    # register subscriber
+    batch_subscriber = subscribe(runs_events_subscriber)
+
+    async def get_subscriber():
+        return batch_subscriber
+
+    # FastStream uses WeakSet for subscribers, so we need to keep long lived reference somewhere
+    dependency_provider.override(BatchSubscriber, get_subscriber)
+
+    # register publisher
     publisher = broker.publisher(settings.producer.malformed_topic)
 
-    # perform registration
-    subscriber(runs_events_subscriber)
+    async def get_publisher():
+        return publisher
+
+    dependency_provider.override(DefaultPublisher, get_publisher)
 
-    dependency_provider.override(AsyncSession, create_session_factory(settings.database))
-    dependency_provider.override(AsyncAPIDefaultPublisher, lambda: publisher)
+    # Override session generator
+    dependency_provider.override(AsyncSession, session_generator(settings.database))
     return broker
 
 
@@ -78,11 +93,13 @@ async def security_lifespan(context: ContextRepo):
                 raise exception from None
 
     return FastStream(
-        broker=broker_factory(settings),
+        broker_factory(settings),
         lifespan=security_lifespan,
-        title="Data.Rentgen",
-        description="Data.Rentgen is a nextgen DataLineage service",
-        version=data_rentgen.__version__,
+        specification=AsyncAPI(
+            title="Data.Rentgen",
+            description="Data.Rentgen is a nextgen DataLineage service",
+            version=data_rentgen.__version__,
+        ),
         logger=logger,
     ).as_asgi(asgi_routes=[("/monitoring/ping", liveness)])
 

data_rentgen/consumer/__main__.py

@@ -8,7 +8,7 @@
 import sys
 from pathlib import Path
 
-from faststream.cli.main import cli
+from faststream.cli import cli
 
 here = Path(__file__).resolve()
 

data_rentgen/consumer/subscribers.py

@@ -5,17 +5,17 @@
 
 import asyncio
 from collections.abc import AsyncGenerator
-from typing import cast
+from typing import Annotated, cast
 
 from aiokafka import ConsumerRecord
 from faststream import Depends, Logger, NoCast
 from faststream.kafka import KafkaMessage
-from faststream.kafka.publisher.asyncapi import AsyncAPIDefaultPublisher
+from faststream.kafka.publisher import DefaultPublisher
 from pydantic import TypeAdapter
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from data_rentgen.consumer.extractors import BatchExtractionResult, BatchExtractor
-from data_rentgen.dependencies import Stub
+from data_rentgen.dependencies.stub import Stub
 from data_rentgen.openlineage.run_event import OpenLineageRunEvent
 from data_rentgen.services.uow import UnitOfWork
 
@@ -30,8 +30,8 @@ async def runs_events_subscriber(
     _events: NoCast[list[OpenLineageRunEvent]],
     batch: KafkaMessage,
     logger: Logger,
-    publisher: AsyncAPIDefaultPublisher = Depends(Stub(AsyncAPIDefaultPublisher)),
-    session: AsyncSession = Depends(Stub(AsyncSession)),
+    publisher: Annotated[DefaultPublisher, Depends(Stub(DefaultPublisher))],
+    session: Annotated[AsyncSession, Depends(Stub(AsyncSession))],
 ):
     message_id = batch.message_id
     correlation_id = batch.correlation_id
@@ -182,7 +182,7 @@ async def report_malformed(
     messages: list[ConsumerRecord],
     message_id: str,
     correlation_id: str,
-    publisher: AsyncAPIDefaultPublisher,
+    publisher: DefaultPublisher,
 ):
     # Return malformed messages back to the broker
     for message in messages:

data_rentgen/http2kafka/__init__.py

@@ -6,9 +6,9 @@
 
 import anyio
 from fastapi import FastAPI
-from faststream._compat import ExceptionGroup
+from faststream._internal._compat import ExceptionGroup
 from faststream.kafka import KafkaBroker
-from faststream.kafka.publisher.asyncapi import AsyncAPIDefaultPublisher
+from faststream.kafka.publisher import DefaultPublisher
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import data_rentgen
@@ -94,7 +94,7 @@ async def get_publisher():
     application.dependency_overrides.update(
         {
             Http2KafkaApplicationSettings: get_settings,
-            AsyncAPIDefaultPublisher: get_publisher,
+            DefaultPublisher: get_publisher,
             AsyncSession: session_generator(settings.database),  # type: ignore[dict-item]
         },
     )

data_rentgen/http2kafka/router/__init__.py

@@ -6,7 +6,7 @@
 
 from asgi_correlation_id import correlation_id
 from fastapi import APIRouter, Body, Depends, Request, Response
-from faststream.kafka.publisher.asyncapi import AsyncAPIDefaultPublisher
+from faststream.kafka.publisher import DefaultPublisher
 
 from data_rentgen.db.models.user import User
 from data_rentgen.dependencies.stub import Stub
@@ -37,7 +37,7 @@
 async def send_events_to_kafka(
     event: Annotated[OpenLineageRunEvent, Body()],
     request: Request,
-    kafka_publisher: Annotated[AsyncAPIDefaultPublisher, Depends(Stub(AsyncAPIDefaultPublisher))],
+    kafka_publisher: Annotated[DefaultPublisher, Depends(Stub(DefaultPublisher))],
     current_user: Annotated[User, Depends(get_user(personal_token_policy=PersonalTokenPolicy.REQUIRE))],
 ):
     body_json_bytes = await request.body()

data_rentgen/logging/presets/colored.yml

@@ -10,12 +10,6 @@ filters:
     (): asgi_correlation_id.CorrelationIdFilter
     uuid_length: 32
     default_value: '-'
-  faststream:
-    (): faststream.log.logging.ExtendedFilter
-    default_context:
-      topic: ''
-      group_id: ''
-    message_id_ln: 10
 
 formatters:
   colored:
@@ -25,7 +19,7 @@ formatters:
     datefmt: '%Y-%m-%d %H:%M:%S'
   kafka_colored:
     (): coloredlogs.ColoredFormatter
-    fmt: '%(asctime)s.%(msecs)03d %(processName)s:%(process)d %(name)s:%(lineno)d [%(levelname)s] %(topic)s %(group_id)s %(message_id)s %(message)s'
+    fmt: '%(asctime)s.%(msecs)03d %(processName)s:%(process)d %(name)s:%(lineno)d [%(levelname)s] %(message)s'
     datefmt: '%Y-%m-%d %H:%M:%S'
 
 handlers:
@@ -37,7 +31,7 @@ handlers:
   faststream:
     class: logging.StreamHandler
     formatter: kafka_colored
-    filters: [faststream]
+    filters: []
     stream: ext://sys.stdout
 
 loggers:

data_rentgen/logging/presets/json.yml

@@ -9,12 +9,6 @@ filters:
     (): asgi_correlation_id.CorrelationIdFilter
     uuid_length: 32
     default_value: '-'
-  faststream:
-    (): faststream.log.logging.ExtendedFilter
-    default_context:
-      topic: ''
-      group_id: ''
-    message_id_ln: 10
 
 formatters:
   json:
@@ -24,7 +18,7 @@ formatters:
     timestamp: true
   kafka_json:
     (): pythonjsonlogger.jsonlogger.JsonFormatter
-    fmt: '%(processName)s:%(process)d %(name)s:%(lineno)d [%(levelname)s] %(topic)s %(group_id)s %(message_id)s %(message)s'
+    fmt: '%(processName)s:%(process)d %(name)s:%(lineno)d [%(levelname)s] %(message)s'
     timestamp: true
 
 handlers:
@@ -36,7 +30,7 @@ handlers:
   faststream:
     class: logging.StreamHandler
     formatter: kafka_json
-    filters: [faststream]
+    filters: []
     stream: ext://sys.stdout
 
 loggers:

data_rentgen/logging/presets/plain.yml

@@ -10,12 +10,6 @@ filters:
     (): asgi_correlation_id.CorrelationIdFilter
     uuid_length: 32
     default_value: '-'
-  faststream:
-    (): faststream.log.logging.ExtendedFilter
-    default_context:
-      topic: ''
-      group_id: ''
-    message_id_ln: 10
 
 formatters:
   plain:
@@ -25,7 +19,7 @@ formatters:
     datefmt: '%Y-%m-%d %H:%M:%S'
   kafka_plain:
     (): logging.Formatter
-    fmt: '%(asctime)s.%(msecs)03d %(processName)s:%(process)d %(name)s:%(lineno)d [%(levelname)s] %(topic)s %(group_id)s %(message_id)s %(message)s'
+    fmt: '%(asctime)s.%(msecs)03d %(processName)s:%(process)d %(name)s:%(lineno)d [%(levelname)s] %(message)s'
     datefmt: '%Y-%m-%d %H:%M:%S'
 
 handlers:
@@ -37,7 +31,7 @@ handlers:
   faststream:
     class: logging.StreamHandler
     formatter: kafka_plain
-    filters: [faststream]
+    filters: []
     stream: ext://sys.stdout
 
 loggers:

pyproject.toml

@@ -75,7 +75,7 @@ server = [
   "cachetools~=6.2.0",
 ]
 consumer = [
-  "faststream[kafka,cli]~=0.5.44",
+  "faststream[kafka,cli]~=0.6.0rc2",
   "cramjam~=2.11.0",
   "pydantic-settings~=2.10.1",
   "alembic~=1.16.2",
@@ -90,8 +90,8 @@ consumer = [
   "packaging~=25.0",
 ]
 http2kafka = [
-  "fastapi>=0.115.14,<0.117.0",
-  "starlette>=0.46.2,<0.48.0",
+  "fastapi~=0.116.1",
+  "starlette~=0.47.2",
   "uvicorn~=0.35.0",
   "starlette-exporter~=0.23.0",
   "asgi-correlation-id~=4.3.4",
@@ -101,7 +101,7 @@ http2kafka = [
   "coloredlogs~=15.0.1",
   "uuid6~=2025.0.0",
   "packaging~=25.0",
-  "faststream[kafka,cli]~=0.5.44",
+  "faststream[kafka,cli]~=0.6.0rc2",
   "cramjam~=2.11.0",
   "cachetools~=6.2.0",
 ]
@@ -155,7 +155,7 @@ docs = [
   # uncomment after https://github.com/zqmillet/sphinx-plantuml/pull/4
   # "sphinx-plantuml~=1.0.0",
   "sphinx-argparse~=0.5.2",
-  "sphinx-tabs>=3.4.7",
+  "sphinx-tabs~=3.4.7",
 ]
 
 [tool.pytest.ini_options]