Building a REST API with Python FastAPI and ClickHouse

Introduction

In this tutorial, we will guide you through the process of creating a REST API using Python FastAPI and ClickHouse. We will connect to a ClickHouse database named coffeecms and create two endpoints:

• api/GetAllUser: This endpoint will display a list of all the records in the users table.
• api/UpdateUser: This endpoint will update the fullname and email fields in the users table for a given user_id.

Prerequisites

Before you begin, you will need to have the following installed on your system:

• Python 3.6 or later
• ClickHouse
• FastAPI

Creating the ClickHouse Database

1. Create a ClickHouse database named coffeecms. You can do this using the following command:

clickhouse-client -q "CREATE DATABASE coffeecms"

2. Create a table named users in the coffeecms database:

CREATE TABLE users (
  user_id UInt64 PRIMARY KEY,
  fullname String NOT NULL,
  email String NOT NULL
) ENGINE = MergeTree()

3. Insert some data into the users table:

INSERT INTO users (user_id, fullname, email) VALUES
  (1, 'John Doe', '[email protected]'),
  (2, 'Jane Doe', '[email protected]'),
  (3, 'Peter Jones', '[email protected]');

Creating the FastAPI Application

1. Create a new directory for your FastAPI application.
2. Create a file named main.py in the directory you created in step 1.
3. Add the following code to the main.py file:

from fastapi import FastAPI
from clickhouse_driver import Client
from pydantic import BaseModel

app = FastAPI()

# Connect to the ClickHouse database
client = Client('localhost')

class User(BaseModel):
  user_id: int
  fullname: str
  email: str

class UserIn(BaseModel):
  fullname: str
  email: str

class UserUpdate(BaseModel):
  fullname: str
  email: str

# Define the GetAllUser endpoint
@app.get("/api/GetAllUser", tags=["Users"])
async def get_all_users():
  # Execute the SQL query to get all users
  query = 'SELECT * FROM users'
  result = client.execute(query)

  # Convert the ClickHouse results to a JSON format
  json_users = []
  for row in result:
    json_user = {
      'user_id': row[0],
      'fullname': row[1],
      'email': row[2]
    }
    json_users.append(json_user)

  # Return the JSON response
  return json_users

# Define the CreateUser endpoint
@app.post("/api/CreateUser", tags=["Users"])
async def create_user(user: UserIn):
  # Create a new user_id
  user_id = client.execute('SELECT max(user_id) FROM users')[0][0] + 1

  # Insert the new user into the database
  query = f"INSERT INTO users (user_id, fullname, email) VALUES ({user_id}, '{user.fullname}', '{user.email}')"
  client.execute(query)

  # Return the new user
  return User(user_id=user_id, fullname=user.fullname, email=user.email)

# Define the UpdateUser endpoint
@app.put("/api/UpdateUser/{user_id}", tags=["Users"])
async def update_user(user_id: int, user: UserUpdate):
  # Update the user's fullname and email
  query = f"UPDATE users SET fullname = '{user.fullname}', email = '{user.email}' WHERE user_id = {user_id}"
  client.execute(query)

  # Return a success message
  return {"message": "User updated successfully"}

# Run the FastAPI application
if __name__ == '__main__':
  import uvicorn
  uvicorn.run(app, host='0.0.0.0', port=8000)

Running the Application

1. Open a terminal window and navigate to the directory where you saved the main.py file.
2. Run the following command to start