Repository navigation
Conversation
- Add HTTPStreamTransport, EventQueue, and get_http_stream_transport() - Implement SSE endpoint for real-time event streaming - Integrate Prometheus metrics for HTTP stream monitoring - Enhance MetricsCollector with step_started() and async-safe singleton - Integrate metrics into PipelineMiddleware and WebSocket transport - Rewrite WebSocket transport with proper auth flow (requires auth message) - Add TransportMiddleware support for http_stream and redis_pubsub types - Add migration guide documentation - Update performance guide with ResourceAwareExecutor examples - Update CI/CD workflow to develop branch with GH_TOKEN - Remove spurious type: ignore comments across codebase - Fix 63 ruff linting errors and type checking issues
dorel14
changed the title
Change the git push target from master to develop in the llms-doc workflow to match the current development branch.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request updates Taskiq-Flow to version 1.0.2, introducing a new HTTP Streaming (SSE) transport, integrating a centralized MetricsCollector for tracking pipeline and WebSocket events, and refining the WebSocket authentication flow. It also includes a comprehensive migration guide and standardizes docstring formatting across the codebase. Feedback focuses on critical logic and security issues: a potential NameError in the WebSocket server due to conditional variable definition, an incorrect header format in the mock Starlette request that would break authentication, and a lack of security checks on the new SSE endpoint. Additionally, the error middleware currently provides inaccurate metadata by using hardcoded retry attempt counts.
| if self.authorization and self.user: | ||
| pipeline_id = channel.replace("pipeline.", "").split(".")[0] | ||
| if not self.authorization.can_read(pipeline_id, self.user): | ||
| transport.send( | ||
| WSMsgType.TEXT, | ||
| json.dumps( | ||
| {"error": "Access denied to pipeline"} | ||
| ).encode(), | ||
| ) | ||
| return | ||
| task = asyncio.create_task( | ||
| self.server.subscribe(channel, transport) | ||
| self.server.subscribe(channel, cast(Any, transport)) | ||
| ) | ||
| task.add_done_callback(lambda t: None) | ||
| self.pipeline_id = channel # For cleanup | ||
| self.pipeline_id = channel | ||
| logger.info("Client subscribed to channel %s", channel) | ||
| MetricsCollector().websocket_message( | ||
| pipeline_id or "unknown", | ||
| "in", | ||
| "subscribe", | ||
| ) |
There was a problem hiding this comment.
The variable pipeline_id is defined conditionally inside the if self.authorization and self.user block but is used unconditionally in the MetricsCollector().websocket_message call. If authorization is not configured, this will raise a NameError. It should be extracted from the channel name before the conditional block.
pipeline_id = channel.replace("pipeline.", "").split(".")[0]
if self.authorization and self.user:
if not self.authorization.can_read(pipeline_id, self.user):
transport.send(
WSMsgType.TEXT,
json.dumps(
{"error": "Access denied to pipeline"}
).encode(),
)
return
task = asyncio.create_task(
self.server.subscribe(channel, cast(Any, transport))
)
task.add_done_callback(lambda t: None)
self.pipeline_id = channel
logger.info("Client subscribed to channel %s", channel)
MetricsCollector().websocket_message(
pipeline_id or "unknown",
"in",
"subscribe",
)| "headers": { | ||
| "authorization": f"Bearer {token}", | ||
| }, |
There was a problem hiding this comment.
The headers field in the Starlette Request scope must be an iterable of tuples containing bytes (e.g., list[tuple[bytes, bytes]]). Providing a dictionary with string keys and values will cause request.headers.get() to fail silently or raise an error when the AuthProvider attempts to verify the token, as Starlette's Headers class expects bytes for comparison.
| "headers": { | |
| "authorization": f"Bearer {token}", | |
| }, | |
| "headers": [ | |
| (b"authorization", f"Bearer {token}".encode()), | |
| ], |
| "Install with: pip install starlette" | ||
| ) from e | ||
|
|
||
| async def sse_endpoint(request: Any) -> "StreamingResponse | None": |
There was a problem hiding this comment.
The sse_endpoint implementation lacks any authentication or authorization checks. While the WebSocket transport properly enforces security via AuthProvider and PipelineAuthorization, this SSE endpoint allows any client to stream pipeline events if they know the pipeline_id. This is a significant security regression since security is now intended to be enabled by default.
| attempt=1, | ||
| max_attempts=1, |
There was a problem hiding this comment.
The attempt and max_attempts fields are hardcoded to 1. This provides inaccurate information to hook subscribers when a task fails during a retry. These values should ideally be extracted from the task message labels if retries are enabled.
Adds unified storage and cache infrastructure with BaseStorageAdapter, InMemoryCacheAdapter, and RedisCacheAdapter. Introduces CacheMiddleware and StorageMiddleware. Adds DAG visualization examples with NetworkX analysis and NiceGUI integration. Upgrades version to 1.2.0.
…tting improvements Adds new API documentation sections for Storage, Cache, and Optimization modules. Updates user guides index with Storage & Cache guide. Cleans up markdown formatting across documentation files by removing bold/italic markers and standardizing table formatting. Maintains consistency between English and French documentation versions.
1 participant
No description provided.