How to Integrate Apache Pulsar with FastAPI
Tags: python, fastapi, apache, pulsar
Published: 2023-07-21 22:56:24

---
title: How to Integrate Apache Pulsar with FastAPI
published: true
date: 2023-07-17 15:49:32 UTC
tags: #python #fastapi #apache #pulsar
canonical_url: https://carlosmv.hashnode.dev/how-to-integrate-apache-pulsar-with-fastapi
---

This is an article about how to set up Apache Pulsar, produce messages and consume them using FastAPI.

This article will not show all the features of Apache Pulsar, or FastAPI.

Apache Pulsar and FastAPI are two powerful technologies that when combined can help you build extremely fast and scalable real-time APIs. Apache Pulsar is an open-source distributed streaming platform that acts as a broker between producers and consumers of data streams. FastAPI is a modern, fast (high-performance), web framework for building APIs with Python. In this article, we will see how to integrate Apache Pulsar with FastAPI to leverage the event streaming capability of Pulsar to power your FastAPI applications.

### Requirements

- Java Runtime Environment (64-bit)

- Python installed

- Pip installed

- WSL installed for Windows users

### Apache Pulsar

Pulsar is a multi-tenant, high-performance solution for server-to-server messaging. Originally developed by Yahoo, Pulsar is under the stewardship of the [Apache Software Foundation](https://www.apache.org/).

Key features of Pulsar are listed below:

- Native support for multiple clusters in a Pulsar instance, with seamless geo-replication of messages across clusters.

- Very low publish and end-to-end latency.

- Seamless scalability to over a million topics.

- A simple client API with bindings for Java, Go, Python and C++.

- Multiple subscription types (exclusive, shared, and failover) for topics.

- Guaranteed message delivery with persistent message storage provided by Apache BookKeeper. A serverless lightweight computing framework Pulsar Functions offers the capability for stream-native data processing.

- A serverless connector framework Pulsar IO, which is built on Pulsar Functions, makes it easier to move data in and out of Apache Pulsar.

- Tiered Storage offloads data from hot/warm storage to cold/long-term storage (such as S3 and GCS) when the data is aging out.

### Installation

To install Apache Pulsar, we go to [this page](https://pulsar.apache.org/download/), download the binary file, and unzip it.

Or execute the following commands:

```
wget https://archive.apache.org/dist/pulsar/pulsar-3.0.0/apache-pulsar-3.0.0-bin.tar.gz


tar xvfz apache-pulsar-3.0.0-bin.tar.gz


cd apache-pulsar-3.0.0
ls -1F

```

Windows users should execute the commands above using WSL.

To start a Pulsar cluster, we run this command to start a standalone Pulsar cluster:

```
bin/pulsar standalone

```

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1689541460990/4ffc22e4-b2f9-4c20-880a-aa362a49563f.png)

After we started our Pulsar cluster, we create two files, a producer and a consumer.

As the Pulsar terminology in the [documentation](https://pulsar.apache.org/docs/3.0.x/reference-terminology/#cluster) says, a producer is a process that publishes messages to a Pulsar topic. A process that establishes a subscription to a Pulsar topic and processes messages published to that topic by producers.

A topic is a named channel used to pass messages published by producers to consumers who process those messages.

**producer.py**

```
import pulsar

client = pulsar.Client('pulsar://localhost:6650')

producer = client.create_producer('my-topic')

for i in range(10):
    producer.send(('Hello-%d' % i).encode('utf-8'))

client.close()

```

In this code, we import the Apache Pulsar client library. Then, we connect to a Pulsar broker running in the localhost:6650. We create a producer, with 'my-topic" as the topic. Then, we create a For loop to send 10 messages to the topic we define. Finally, we close the connection.

**consumer.py**

```
import pulsar

client = pulsar.Client('pulsar://localhost:6650')

consumer = client.subscribe('my-topic', 'my-subscription')

while True:
    msg = consumer.receive()
    try:
        print("Received message '{}' id='{}'".format(msg.data(), msg.message_id()))
        # Acknowledge successful processing of the message
        consumer.acknowledge(msg)
    except Exception:
        # Message failed to be processed
        consumer.negative_acknowledge(msg)

client.close()

```

In the consumer, we import the Apache Pulsar client library and create a connection to the Pulsar broker running locally on port 6650. We subscribe to the 'my-topic' topic with the subscription name 'my-subscription'.

We create an infinite loop that will continuously receive and process messages. If successful, it acknowledges the message. If an exception occurs, it negatively acknowledges the message.

Then, we close the Pulsar client and connection.

We run in different terminals the `producer.py` file and the `consumer.py` file.

This is the `producer.py` file terminal output:

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1689541351336/57329cc9-21cb-4485-b5d6-004ed73ea95d.png)

And this is the `consumer.py` file terminal output:

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1689541384771/7c906372-c5ad-4767-8417-b922c002e79e.png)

### Creating a FastAPI server

Now, we are going to create a server that sends log information to the consumer.

Every time a request is made to the server, the server will send the logs to the "logs" topic.

**main.py**

```
from fastapi import FastAPI, Request
from datetime import datetime
import pulsar 

app = FastAPI()

@app.get("/")
async def root(request: Request):

    ip_address = request.client.host
    request_url = request.url._url
    request_port = request.url.port
    request_path = request.url.path
    request_method = request.method
    request_time = datetime.now()
    browser_type = request.headers["User-Agent"]
    operating_system = request.headers["Accept"]
    message = {
        "ip_address": ip_address,
        "request_url": request_url,
        "request_port": request_port,
        "request_path": request_path,
        "request_method": request_method,
        "request_time": request_time,
        "browser_type": browser_type,
        "operating_system": operating_system,
    }

    client = pulsar.Client('pulsar://localhost:6650')
    producer = client.create_producer('my-topic')

    producer.send(str(message).encode("utf-8"))
    client.close()
    return {"message": message}

if __name__ == " __main__":
    app.run(debug=True)

```

In the `main.py` file, we import the FastAPI, Request, datetime and pulsar libraries.

We create a FastAPI app and define a root endpoint `/` that receives a Request object. And extract various information from the request like IP address, URL, method, headers etc. Then, we create a message dictionary with the extracted request information.

We create a connection to a Pulsar broker, create a producer and send the message dictionary to the 'my-topic' topic.

And finally, this code returns the message dictionary in the response.

We convert the Python dictionary to a string to avoid this message error:

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1689538307077/af785d51-b24d-41fd-9e82-f8bb7bafc4aa.png)

If we navigate to `locahost:8000` with a web browser, we should see this response.

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1689538454456/8b4d4a23-0c6e-48fa-ac2f-b06d4950ec59.png)

We could use an HTTP client, and we should receive the same response.

We run the `consumer.py` file in another terminal. And should receive the following response:

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1689538328570/9ca97b83-c33d-4301-8c50-2646f42bbdcf.png)

### Conclusion

Integrating Apache Pulsar with FastAPI enables you to build high-performance real-time APIs. Pulsar handles the complexities of event streaming and event storage while FastAPI provides a simple yet powerful framework to build the APIs. This combination allows you to decouple the event producer from the API consumer.

Thank you for taking the time to read this article.

If you have any recommendations about other packages, architectures, how to improve my code, my English, or anything; please leave a comment or contact me through [**Twitter**](https://twitter.com/Carlos_marcv), or [**LinkedIn**](https://www.linkedin.com/in/carlos-marcano-a2135a134/).

The source code is [here](https://github.com/carlosm27/fastapi_pulsar_demo).

### Resources

[Run a standalone Pulsar cluster locally](https://pulsar.apache.org/docs/3.0.x/getting-started-standalone/)

[Set up Python client](https://pulsar.apache.org/docs/3.0.x/client-libraries-python-setup/)

[Pulsar Terminology](https://pulsar.apache.org/docs/3.0.x/reference-terminology/#cluster)

[Use Python client](https://pulsar.apache.org/docs/3.0.x/client-libraries-python-use/)

[Produce and consume messages](https://pulsar.apache.org/docs/3.0.x/tutorials-produce-consume/)