fix: Correct type hinting for events retrieval in service-adapters
Some checks failed
Integration Tests / integration-tests (push) Failing after 28s
Integration Tests / performance-tests (push) Has been skipped
Docker Build and Push / build-and-push (push) Failing after 32s
Service Adapters (Python FastAPI) / test (3.11) (push) Failing after 39s
Service Adapters (Python FastAPI) / test (3.14) (push) Failing after 10s
Service Adapters (Python FastAPI) / test (3.12) (push) Failing after 42s
Service Adapters (Python FastAPI) / test (3.13) (push) Failing after 40s
Service Adapters (Python FastAPI) / build (push) Has been skipped
Some checks failed
Integration Tests / integration-tests (push) Failing after 28s
Integration Tests / performance-tests (push) Has been skipped
Docker Build and Push / build-and-push (push) Failing after 32s
Service Adapters (Python FastAPI) / test (3.11) (push) Failing after 39s
Service Adapters (Python FastAPI) / test (3.14) (push) Failing after 10s
Service Adapters (Python FastAPI) / test (3.12) (push) Failing after 42s
Service Adapters (Python FastAPI) / test (3.13) (push) Failing after 40s
Service Adapters (Python FastAPI) / build (push) Has been skipped
### Summary of Changes - Updated the type hinting in `events.py` to use `cast` for the list of events retrieved from Redis, ensuring type safety and clarity in the code. ### Expected Results - Improved type checking and maintainability of the service-adapters module, enhancing overall code quality.
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
import json
|
import json
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import List
|
from typing import List, cast
|
||||||
|
|
||||||
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query
|
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query
|
||||||
|
|
||||||
@@ -53,7 +53,7 @@ async def publish_event(event_data: EventData, background_tasks: BackgroundTasks
|
|||||||
async def get_events(limit: int = Query(100, ge=1, le=1000, description="Maximum number of events to retrieve")):
|
async def get_events(limit: int = Query(100, ge=1, le=1000, description="Maximum number of events to retrieve")):
|
||||||
"""Get recent events from the Redis message bus"""
|
"""Get recent events from the Redis message bus"""
|
||||||
try:
|
try:
|
||||||
events: List[str] = redis_client.lrange("events", 0, limit - 1)
|
events: List[str] = cast(List[str], redis_client.lrange("events", 0, limit - 1))
|
||||||
parsed_events = []
|
parsed_events = []
|
||||||
for event in events:
|
for event in events:
|
||||||
try:
|
try:
|
||||||
|
|||||||
Reference in New Issue
Block a user