Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use get_publisher_client method in testcontainers-python

Best Python code snippet using testcontainers-python_python

pub_sub.py

Source:pub_sub.py

2import google.api_core.exceptions3from google.api_core import retry4from google.cloud import pubsub_v15from google.oauth2 import service_account6def get_publisher_client():7    return pubsub_v1.PublisherClient()8def get_subscription_client():9    return pubsub_v1.SubscriberClient()10def create_topic(project_id, topic_id):11    publisher = get_publisher_client()12    topic_path = publisher.topic_path(project_id, topic_id)13    try:14        topic = publisher.create_topic(request={"name": topic_path})15        return {"success": "Topic created successfully"}16    except google.api_core.exceptions.AlreadyExists:17        return {"success": "Topic already exists"}18def create_subscription(project_id, topic_id, subscription_id):19    publisher = get_publisher_client()20    subscriber = get_subscription_client()21    topic_path = publisher.topic_path(project_id, topic_id)22    subscription_path = subscriber.subscription_path(project_id, subscription_id)23    try:24        with subscriber:25            subscription = subscriber.create_subscription(26                request={"name": subscription_path, "topic": topic_path}27            )28        return {"success": "Subscription created successfully"}29    except google.api_core.exceptions.AlreadyExists:30        return {"success": "Subscription already exists"}31def publish_messages(project_id, topic_id, message):32    publisher = get_publisher_client()33    topic_path = publisher.topic_path(project_id, topic_id)34    try:35        future = publisher.publish(topic_path, message.encode("utf-8"))36        return {"success": f"Published messages to {topic_path}"}37    except Exception as e:38        return {"error": "Internal Server Error"}39def pull_messages(project_id, subscription_id, num_messages):40    subscriber = get_subscription_client()41    subscription_path = subscriber.subscription_path(project_id, subscription_id)42    with subscriber:43        response = subscriber.pull(44            request={"subscription": subscription_path, "max_messages": num_messages},45            retry=retry.Retry(deadline=300),46        )...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.