gcp-pubsub-events 2.0.0

A decorator-based library for handling Google Cloud Pub/Sub messages with FastAPI integration, inspired by Micronaut's @PubSubListener

GCP PubSub Events

A modern, decorator-based Python library for Google Cloud Pub/Sub event handling. Inspired by event-driven architectures and frameworks like Micronaut, this library makes it easy to build scalable, event-driven microservices with minimal boilerplate.

Features

• Decorator-based API: Simple @pubsub_listener and @subscription decorators
• Type Safety: Full Pydantic model support with automatic validation
• Async Support: Works with both sync and async message handlers
• Framework Integration: Works seamlessly with FastAPI, Flask, Django
• Auto-retry Logic: Built-in retry mechanism with ack/nack support
• Resource Management: Optional automatic topic and subscription creation
• Comprehensive Testing: Well-tested with unit, integration, and e2e tests

Installation

pip install gcp-pubsub-events

Core Concepts

Decorators

• @pubsub_listener: Marks a class as containing event handlers
• @subscription("subscription-name"): Marks a method as handling messages from a specific subscription

Message Acknowledgment

Every handler receives an Acknowledgement object:

• ack.ack(): Message processed successfully, remove from queue
• ack.nack(): Processing failed, retry later

Context Managers

The library provides context managers for lifecycle management:

• pubsub_manager: Synchronous context manager
• async_pubsub_manager: Asynchronous context manager for async frameworks

Basic Usage

from gcp_pubsub_events import pubsub_listener, subscription, PubSubManager

@pubsub_listener
class EventHandler:
    @subscription("my-subscription")
    def handle_message(self, data: dict, ack):
        # Process the message
        print(f"Received: {data}")
        ack.ack()

# Create handler instance
handler = EventHandler()

# Use context manager for lifecycle management
with PubSubManager("my-project-id") as manager:
    # Application runs here
    # Manager handles all setup and cleanup
    pass

FastAPI Integration

from contextlib import asynccontextmanager
from fastapi import FastAPI
from gcp_pubsub_events import async_pubsub_manager, pubsub_listener, subscription

@pubsub_listener
class MyService:
    @subscription("my-subscription")
    async def handle_event(self, data: dict, ack):
        # Process event
        ack.ack()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create service instance first
    service = MyService()
    
    # Then start the manager
    async with async_pubsub_manager("my-project") as manager:
        yield

app = FastAPI(lifespan=lifespan)

Configuration

Environment Variables

# Required
export GOOGLE_CLOUD_PROJECT="my-project-id"

# Optional
export PUBSUB_EMULATOR_HOST="localhost:8085"  # For local development
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account.json"

Manager Configuration

from gcp_pubsub_events import PubSubManager

manager = PubSubManager(
    project_id="my-project",
    max_workers=10,              # Thread pool size
    max_messages=100,            # Max concurrent messages
    auto_create_resources=True,  # Auto-create topics/subscriptions
    clear_registry_on_start=False # Clear handlers on restart
)

Testing

The library includes comprehensive test support:

# Run all tests
poetry run pytest

# Run with emulator
export PUBSUB_EMULATOR_HOST=localhost:8085
gcloud beta emulators pubsub start
poetry run pytest tests/integration/

Contributing

We welcome contributions! Please see our Contributing Guide for details.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Resources

• Google Cloud Pub/Sub Documentation
• Issue Tracker