DOCSEARCHbyAlgolia
Submit Docs

python-pubsub

repository·main·Indexed 19 days ago

https://github.com/googleapis/python-pubsub

A Python client library for Google Cloud Pub/Sub, a fully-managed real-time messaging service used to decouple applications via topics and subscriptions. The library provides the PublisherClient for sending messages with configurable batching and flow control, and the SubscriberClient for pulling messages synchronously or asynchronously via callbacks.

Tokens
26.4K
Snippets
82
Records
102
Agent score
65%

What's inside python-pubsub

  1. Understand Acknowledge (ack) and Negative Acknowledge (nack)

    main

    In Pub/Sub, managing message lifecycle is critical for reliability:

    • Ack (Acknowledge): Call message.ack() when processing is complete. This tells Pub/Sub the message does not need to be sent again.
    • Nack (Negative Acknowledge): Call message.nack() when you are unable or unwilling to process a message. This tells Pub/Sub to redeliver the message.

    Best Practice: Avoid acknowledging messages immediately upon receipt. If your processing fails after an early ack, the message is lost. Only ack after successful processing so that Pub/Sub can redeliver unacknowledged messages in case of failure.

  2. Understand the SchemaServiceTransport inheritance structure

    main

    The SchemaServiceTransport serves as the Abstract Base Class (ABC) for all transport implementations used by the Schema Service. Depending on your application's concurrency model (synchronous vs. asynchronous) and preferred protocol (gRPC vs. REST), you should use one of the following public transport classes:

    • gRPC (Synchronous): Use SchemaServiceGrpcTransport (defined in grpc.py).
    • gRPC (Asynchronous): Use SchemaServiceGrpcAsyncIOTransport (defined in grpc_asyncio.py).
    • REST (Synchronous): Use SchemaServiceRestTransport (defined in rest.py).

    Note that _BaseSchemaServiceRestTransport is a private base class for REST implementations and should not be used directly.

  3. Understand the SubscriberTransport inheritance structure

    main

    The SubscriberTransport class serves as the Abstract Base Class (ABC) for all subscriber transport implementations. Depending on your concurrency model (synchronous vs. asynchronous) and protocol preference (gRPC vs. REST), you will use different child implementations:

    • gRPC (Synchronous): Use SubscriberGrpcTransport (defined in grpc.py).
    • gRPC (Asynchronous): Use SubscriberGrpcAsyncIOTransport (defined in grpc_asyncio.py).
    • REST (Synchronous): Use SubscriberRestTransport (defined in rest.py).

    Note that REST transports utilize a private base class _BaseSubscriberRestTransport and internal METHOD classes to manage API calls.

  4. Use the `request` object in v2.0.0+ method calls

    main

    In version 2.0.0 and later, most methods that interact with the backend use a single positional parameter named request.

    Important Rules:

    • The request parameter and flattened keyword parameters are mutually exclusive. You cannot pass both in the same call.
    • Some methods require specific options that are only available within the request object.
    • Exceptions: Hand-written methods like publisher.publish() and subscriber.subscribe() (which handle logic like batching) have largely preserved their original signatures.

    Valid Patterns:

    # Using a request dictionary
response = client.list_topics(request={"project": project_path})

# Using keyword arguments (alternative to request)
response = client.list_topics(project=project_path)

    Invalid Pattern (will cause an error):

    # Mixing request and keyword arguments
response = client.list_topics(request={"project": project_path}, metadata=[("foo", "bar")])
    # Before 2.0.0
topics = publisher.list_topics(project_path)

# After 2.0.0
topics = publisher.list_topics(request={"project": project_path})

  5. Understand the PublisherTransport inheritance structure

    main

    The PublisherTransport Abstract Base Class (ABC) serves as the foundation for all transport implementations in the publisher service. Depending on your concurrency model (synchronous vs. asynchronous) and protocol (gRPC vs. REST), you should choose the appropriate child class:

    • gRPC (Synchronous): Use PublisherGrpcTransport (defined in grpc.py).
    • gRPC (Asynchronous): Use PublisherGrpcAsyncIOTransport (defined in grpc_asyncio.py).
    • REST (Synchronous): Use PublisherRestTransport (defined in rest.py). This class uses METHOD inner classes derived from the private _BaseMETHOD classes.
    • REST (Base/Internal): _BasePublisherRestTransport (defined in rest_base.py) acts as the private base for REST implementations.

  6. Handle publish results with Futures

    main

    The publish method returns a Future object. You can use this to verify if a message was successfully published or to handle results asynchronously.

    Synchronous approach: Call .result() on the future. This blocks the execution until the publish request is complete. If the publish fails, .result() will raise an exception.

    Asynchronous approach (Callbacks): Use .add_done_callback(callback) to attach a function that executes once the future is complete. The callback receives the future object as its only argument. If the future is already finished when the callback is added, it is executed immediately.

    # Synchronous: blocking until complete
future = client.publish(topic, b'My awesome message.')
message_id = future.result()

# Asynchronous: using a callback
def callback(future):
    message_id = future.result()
    do_something_with(message_id)

future = client.publish(topic, b'My awesome message.')
future.add_done_callback(callback)

  7. Pull messages asynchronously with callbacks

    main

    The subscriber.subscribe method starts a background thread to receive messages and executes a callback function for each message. This method returns a StreamingPullFuture which can be used to manage the background thread.

    # Define the callback.
# Note that the callback is defined *before* the subscription is opened.
def callback(message):
    do_something_with(message)  # Replace this with your actual logic.
    message.ack()  # Asynchronously acknowledge the message.

# Wrap the following code in `with pubsub.SubscriberClient() as subscriber:`
# Substitute PROJECT and SUBSCRIPTION with appropriate values.

subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION)

# Open the subscription, passing the callback.
future = subscriber.subscribe(subscription_path, callback)

  8. Quickstart: Publish and Subscribe messages

    main

    Use these minimal snippets to quickly test your Pub/Sub setup by publishing a message to a topic and receiving it via a subscription.

    To publish a message: Run the publisher quickstart script:

    python3 quickstart/pub.py

    To subscribe to messages: Run the subscriber quickstart script:

    python3 quickstart/sub.py
    $ python3 quickstart/pub.py
$ python3 quickstart/sub.py

  9. Migrate to google-cloud-pubsub v2.0.0

    main

    The 2.0.0 release of google-cloud-pubsub introduced significant breaking changes due to a new code generator. Key changes include:

    • Python Version: Requires Python 3.6+.
    • Method Signatures: Most methods now expect a single request object instead of positional arguments.
    • Removed Methods: Utility methods like subscription_path() have been moved to their specific clients (e.g., subscriber.subscription_path()), and project_path() has been removed entirely.
    • Client Configuration: The client_config parameter is no longer supported during client construction.

    To automate the migration of common method calls, use the provided fixup_pubsub_v1_keywords.py script.

  11. Install google-cloud-pubsub

    main

    Install the library in a virtualenv to avoid dependency conflicts.

    Supported Python Versions:

    • Python >= 3.9
    • For Python 3.7 and 3.8, use google-cloud-pubsub==2.34.0.
    • For Python 2.7, use google-cloud-pubsub==1.7.0.

    Note: This repository is archived. The project has moved to google-cloud-python.

    ### Mac/Linux
```bash
pip install virtualenv
virtualenv <your-env>
source <your-env>/bin/activate
<your-env>/bin/pip install google-cloud-pubsub

    Windows

    pip install virtualenv
virtualenv <your-env>
<your-env>\Scripts\activate
<your-env>\Scripts\pip.exe install google-cloud-pubsub

  12. Set up a development environment for Pub/Sub samples

    main

    To run the Google Cloud Pub/Sub Python samples, follow these steps to clone the sample repository and set up a Python virtual environment:

    1. Clone the samples repository: Clone python-docs-samples and navigate to the specific Pub/Sub sample directory.
    2. Prepare Python tools: Ensure pip and virtualenv are installed.
    3. Create and activate a virtual environment: Samples are compatible with Python 3.9+.
    4. Install dependencies: Use pip to install the requirements listed in the sample directory.
    # Clone the repository
$ git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git

# Create and activate a virtual environment (Python 3.9+ recommended)
$ virtualenv env
$ source env/bin/activate

# Install dependencies
$ pip install -r requirements.txt