Celes Renata
40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
"""Helpers for enqueuing lake publish jobs from upstream workers.
|
|
|
|
Other services import these helpers to push jobs onto the QUEUE_LAKE_PUBLISH
|
|
Redis queue. The lake publisher worker (jobs.py) consumes them.
|
|
|
|
Usage:
|
|
await enqueue_lake_job(rds, "document", document_id)
|
|
await enqueue_lake_job(rds, "trade_order", order_id)
|
|
await enqueue_lake_job(rds, "bulk_documents", since=cutoff.isoformat())
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import redis.asyncio as aioredis
|
|
|
|
from services.shared.redis_keys import QUEUE_LAKE_PUBLISH, queue_key
|
|
|
|
|
|
async def enqueue_lake_job(
|
|
rds: aioredis.Redis,
|
|
job_type: str,
|
|
entity_id: str = "",
|
|
since: str | None = None,
|
|
) -> None:
|
|
"""Push a lake publish job onto the Redis queue.
|
|
|
|
Args:
|
|
rds: Async Redis client.
|
|
job_type: One of the supported job types (document, document_extraction,
|
|
market_snapshot, trade_order, trade_fill, positions_snapshot,
|
|
pnl_snapshot, bulk_documents, bulk_extractions).
|
|
entity_id: UUID or identifier for the entity to publish.
|
|
since: ISO datetime string for bulk jobs (cutoff timestamp).
|
|
"""
|
|
payload: dict[str, str] = {"job_type": job_type, "entity_id": entity_id}
|
|
if since:
|
|
payload["since"] = since
|
|
await rds.rpush(queue_key(QUEUE_LAKE_PUBLISH), json.dumps(payload)) # type: ignore[misc]
|