Metadata-Version: 2.5
Name: gcp-pubsub-dao
Version: 0.6.1
Summary: DAO for GCP PubSub service.
Author-email: Raman Shakun <shakunroman@gmail.com>
Requires-Python: <4.0,>=3.11
Requires-Dist: google-cloud-pubsub<3.0.0,>=2.19.0
Description-Content-Type: text/markdown

# GCP Pub/Sub DAO
The library provides DAO classes for GCP pubsub publisher/subscriber.

## Installation

```python
pip install gcp-pubsub-dao
```

## Usage

- sync subscriber:
```python
from gcp_pubsub_dao import PubSubSubscriberDAO, Message

dao = PubSubSubscriberDAO(project_id="prodect-dev", subscription_id="subscription")
messages: Message = dao.get_messages(messages_count=2)

for message in messages:
    print(message.data)
    
dao.ack_messages(ack_ids=[message[0].ack_id])      
dao.nack_messages(ack_ids=[message[1].ack_id])     

dao.close()     # to clean up connections
```
- sync publisher:
```python
from gcp_pubsub_dao import PubSubPublisherDAO

dao = PubSubPublisherDAO(project_id="prodect-dev")
try:
    dao.publish_message(topic_name="topic", payload=b"asdfsdf", attributes={"kitId": "AW12345678"})
except Exception as ex:
    print(ex)
```
- async subscriber:
```python
from gcp_pubsub_dao import AsyncPubSubSubscriberDAO, Message

dao = AsyncPubSubSubscriberDAO(project_id="prodect-dev", subscription_id="subscription")
messages: Message = await dao.get_messages(messages_count=2)

for message in messages:
    print(message.data)
    
await dao.ack_messages(ack_ids=[message[0].ack_id])      
await dao.nack_messages(ack_ids=[message[1].ack_id])
```
- async publisher:
```python
from gcp_pubsub_dao import AsyncPubSubPublisherDAO

dao = AsyncPubSubPublisherDAO(project_id="prodect-dev")
try:
    await dao.publish_message(topic_name="topic", payload=b"asdfsdf", attributes={"kitId": "AW12345678"})
except Exception as ex:
    print(ex)
```
- async worker pool

```python
import asyncio
import sys

sys.path.append("./")

from gcp_pubsub_dao import AsyncPubSubSubscriberDAO
from gcp_pubsub_dao.worker_pool import WorkerPool, WorkerTask, HandlerResult
from gcp_pubsub_dao.entities import Message


async def handler1(message: Message):
    print(f"handler1: {message}")
    await asyncio.sleep(2)
    return HandlerResult(ack_id=message.ack_id, is_success=True)


async def handler2(message: Message):
    print(f"handler2: {message}")
    await asyncio.sleep(5)
    return HandlerResult(ack_id=message.ack_id, is_success=True)


def heartbeat_func():
    print("Heartbeat: Worker is alive")


async def main():
    tasks = [
        WorkerTask(
            subscriber_dao=AsyncPubSubSubscriberDAO(project_id="ash-dev-273120", subscription_id="http-sender-sub"),
            handler=handler1,
        ),
        WorkerTask(
            subscriber_dao=AsyncPubSubSubscriberDAO(project_id="ash-dev-273120", subscription_id="email-sender-sub"),
            handler=handler2,
        ),
    ]
    
    # Create worker pool with heartbeat function
    wp = WorkerPool(heartbeat_func=heartbeat_func)
    
    # Run in async mode (default) - all tasks run concurrently
    await wp.run(tasks=tasks)
    
    # Or run in sync mode - tasks run one by one in order
    # await wp.run(tasks=tasks, mode="sync")


if __name__ == "__main__":
    asyncio.run(main())
```

## Worker Pool Features

The `WorkerPool` provides two execution modes:

### Async Mode (default)
- All tasks run concurrently using `asyncio.TaskGroup`
- Tasks can execute in any order or simultaneously
- Best for independent tasks that don't need to be processed in sequence

### Sync Mode
- Tasks run one by one in the order they are provided
- Each task completes before the next one starts
- Useful when tasks need to be processed in a specific sequence
- Note: Message processing within each task is still asynchronous

### Heartbeat Function
- Optional callback function that gets called during worker execution
- Useful for monitoring worker health and activity
- Called before processing messages in each iteration
- Can be used for logging, metrics, or health checks

### WorkerTask Configuration
- `subscriber_dao`: The async subscriber DAO instance
- `handler`: Async function that processes messages and returns `HandlerResult`
- `batch_size`: Number of messages to fetch per batch (default: 10)
- `return_immediately`: Whether to return immediately if no messages (default: False)

## Event Dispatcher

`EventDispatcher` lets a consumer run **one** worker against a single (unfiltered) subscription and route
each message to a handler by event name, instead of running one worker per filtered subscription.

The event name is read from the `event` message attribute and looked up in the handler registry:

| Case | Behavior |
|------|----------|
| Event has a registered handler | The handler is awaited and its `HandlerResult` is returned unchanged — ack on success, nack on failure, same as a plain `WorkerTask` handler. Exceptions propagate to the worker pool as before. |
| Event has no registered handler (or the attribute is missing) | The message is **acked immediately and silently** — no log, no retry, no dead-lettering. |

The dispatcher is itself a handler, so it plugs straight into `WorkerTask`:

```python
from gcp_pubsub_dao import AsyncPubSubSubscriberDAO, EventDispatcher, WorkerPool, WorkerTask

dispatcher = EventDispatcher(
    handlers={
        "kit-accessioned": kit_accessioned_handler.handle,
        "kit-issue": kit_issue_handler.handle,
    },
)

task = WorkerTask(
    subscriber_dao=AsyncPubSubSubscriberDAO(project_id="ash-dev-273120", subscription_id="order-task-my-service"),
    handler=dispatcher,
)
await WorkerPool().run(tasks=[task])
```

### Options

- `handlers`: Mapping of event name → async handler (`Callable[[Message], Awaitable[HandlerResult]]`).
  The mapping is copied on init, so later changes to the original dict have no effect.
- `event_attribute`: Message attribute holding the event name (default: `"event"`).

### Shadow mode

`ShadowModeEventHandler` is a handler that does no real work: it logs the kit id, event name and message id
at `INFO` level and acks the message. Register it in the dispatcher **in place of** the real handlers to
validate a new subscription's routing against live traffic without processing anything:

```python
from loguru import logger

from gcp_pubsub_dao import EventDispatcher, ShadowModeEventHandler

shadow_handler = ShadowModeEventHandler(logger)
dispatcher = EventDispatcher(handlers={event: shadow_handler for event in real_handlers})
```

Each message routed to it produces a single `INFO` record with the context passed as kwargs
(with loguru they end up in `record["extra"]`):

```python
logger.info("Shadow dispatch", kit_id="AW12345678", event="kit-accessioned", message_id="1234567890")
```

Messages that have no entry in the registry never reach the shadow handler, so they are acked silently
and not logged — exactly as they would be once real handlers are in place.

Options:

- `logger`: Any object with `info(msg, **kwargs)` and `warning(msg, **kwargs)` methods accepting arbitrary
  keyword context — e.g. `loguru.logger`. The library does not depend on loguru. Note that the stdlib
  `logging.Logger` does not accept arbitrary kwargs, so wrap it in an adapter if you need to use it.
- `event_attribute`: Message attribute holding the event name (default: `"event"`).
- `kit_id_getter`: Callable extracting the kit id from a message. By default it uses the `kitId`
  attribute, then the `kit_id` field of the JSON payload, otherwise `None`. If the getter raises, a
  warning is logged (with `error` and `message_id` kwargs), `kit_id=None` is reported and the message is
  still acked.
