API Versioning with FastAPI and Neon Branching

During new feature development, we frequently introduce new API versions to roll out new features gradually while keeping the existing API functional for the connected client applications. Oftentimes, the new API version has a new database schema change and we need to serve different datasets based on the API version. In such scenarios, we can use Neon’s branching feature to dynamically manage database versions instead of creating separate databases or handling costly data migrations. This allows us to keep multiple versions of an API with different data structures.

In this guide, we will:

1. Use Neon's API to create and manage database branches dynamically.
2. Implement a FastAPI backend service that automatically connects to different database branches.

You can also try out the example on the GitHub repository.

Use Case: Versioned APIs with Dynamic Databases

Let's say we have an API that manages user data. The v1 API only has id, name, and email, in the User table while the v2 API introduces additional fields like age and city. We use Neon database branching to create different database versions dynamically. We can keep each API version isolated without modifying production data.

This means that:

• API v1 can connect to one branch of the database.
• API v2 can connect to another branch with an updated schema.

Version	Columns
v1	id, name, email
v2	id, name, email, age, city

Lets try to implement this simple project.

Step-by-Step API Versioning Implementation

Prerequisites

Before we begin, make sure you have the following:

• Python 3.8 or later version
• A free Neon account

Create a Neon Project

1. Navigate to the Neon Console
2. Click "New Project"
3. Select Azure as your cloud provider
4. Choose East US 2 as your region
5. Give your project a name (e.g., "api-versioning-neondb")
6. Click "Create Project"
7. Once the project is created successfully, copy the Project ID from Settings under the project settings view.
8. Retrieve the Neon API Key: Create a new API Key, copy it, and save it safely. We will use it in the project.

Set Up FastAPI Project in Python

Project Structure

The final project structure looks like this:

api-versioning-with-neondb-branching/
│
├── app/
│   ├── main.py                   
│   ├── neon_db_setup.py          
|   ├── db_connection.py         
│── data/
│   ├── schema_v1.sql              # SQL schema for v1
│   ├── schema_v2.sql              # SQL schema for v2
│── .env
|── requirements.txt 
└── README.md

Set Up Environment Variables

Create a .env file in the project root directory:

NEON_API_KEY=your_neon_api_key
NEON_PROJECT_ID=your_neon_project_id

Add Python Dependencies

Lists Python dependencies in requirements.txt file:

uvicorn==0.34.0
fastapi==0.115.8
requests==2.32.3
psycopg2-binary==2.9.10
python-dotenv==1.0.1

Initialize Database Schema

Define Schema for API v1:

-- schema_v1.sql
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name TEXT,
    email TEXT
);

INSERT INTO users (name, email) VALUES 
('Alice', '[e-mail]'), 
('Bob', '[e-mail]');

Define Schema for API v2:

-- schema_v2.sql with additional fields
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name TEXT,
    email TEXT,
    age INT,
    city TEXT
);

INSERT INTO users (name, email, age, city) VALUES 
('Alice', '[e-mail]', 25, 'New York'), 
('Bob', '[e-mail]', 30, 'San Francisco');

Install virtual environment (Preferred)

python3 -m venv env
source env/bin/activate

Install the required dependencies

pip install -r requirements.txt

Managing Neon Database Branches

We need to create a new branch for each API version. This is done using Neon's API programmatically.

The below neon_db_setup.py script does the following things:

• Checks if a branch already exists. Creates a new branch for the API version if it doesn't.
• Fetches connection strings for new branches.
• Initializes schema automatically per branch and populates branch databases with sample data by running SQL queries we specified in the data folder.

Everything happens at project startup time and Neon creates new branches instantly.

import os
import requests
import psycopg2
from dotenv import load_dotenv

load_dotenv()

NEON_API_KEY = os.getenv("NEON_API_KEY")
NEON_PROJECT_ID = os.getenv("NEON_PROJECT_ID")
NEON_API_URL = "https://console.neon.tech/api/v2"
DEFAULT_DATABASE_NAME = "neondb"
DEFAULT_DATABASE_ROLE = "neondb_owner"

SCHEMA_FILES = {
    "v1": "./data/schema_v1.sql",
    "v2": "./data/schema_v2.sql",
}

versions = ["v1", "v2"]

def create_or_get_branch(branch_name):
    """
    Checks if a branch exists. If not, creates it.

    :param branch_name: The branch to check or create.
    :return: The branch ID.
    """
    existing_branches = list_existing_branches()

    if branch_name in existing_branches:
        print(f"✅ Branch '{branch_name}' already exists.")
        return existing_branches[branch_name]

    url = f"{NEON_API_URL}/projects/{NEON_PROJECT_ID}/branches"
    headers = {
        "Authorization": f"Bearer {NEON_API_KEY}",
        "Accept": "application/json",
        "Content-type": "application/json",
    }
    data = {"branch": {"name": branch_name}, "endpoints": [{"type": "read_write"}]}

    response = requests.post(url, headers=headers, json=data)

    if response.status_code == 201:
        branch_data = response.json()
        branch_id = branch_data.get("branch", {}).get("id")
        print(f"✅ Branch '{branch_name}' created successfully with ID: {branch_id}")
        return branch_id
    else:
        print(f"❌ Failed to create branch: {response.text}")
        return None

def list_existing_branches():
    """
    Fetches the list of existing branches in the Neon project.

    :return: List of existing branch names.
    """
    url = f"{NEON_API_URL}/projects/{NEON_PROJECT_ID}/branches"
    headers = {"Authorization": f"Bearer {NEON_API_KEY}"}

    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        branches = response.json().get("branches", [])
        return {branch["name"]: branch["id"] for branch in branches}
    else:
        print(f"❌ Failed to fetch existing branches: {response.text}")
        return {}

def get_connection_uri(branch_id):
    """
    Fetches the database connection URI for a specific Neon branch.

    :param branch_id: The ID of the branch.
    :return: The connection string URI.
    """
    url = f"{NEON_API_URL}/projects/{NEON_PROJECT_ID}/connection_uri"
    params = {
        "database_name": DEFAULT_DATABASE_NAME,
        "role_name": DEFAULT_DATABASE_ROLE,
        "branch_id": branch_id,  # Pass the branch ID as a query parameter
    }
    headers = {"Authorization": f"Bearer {NEON_API_KEY}"}

    response = requests.get(url, headers=headers, params=params)

    if response.status_code == 200:
        return response.json().get("uri")
    else:
        print(f"❌ Failed to fetch connection URI: {response.text}")
        return None

def initialize_connection_strings():
    """
    Initializes and caches database connection strings for different API versions.
    This function is called once at application startup.
    """
    connection_strings = {}

    for version in versions:
        branch_name = f"api_{version}_branch"
        branch_id = create_or_get_branch(branch_name)

        if branch_id:
            connection_uri = get_connection_uri(branch_id)
            if connection_uri:
                connection_strings[version] = connection_uri
                print(f"✅ Cached connection for {version}: {connection_uri}")

                # Initialize database schema for this version
                schema_file = SCHEMA_FILES.get(version)
                if schema_file:
                    initialize_database_schema(connection_uri, schema_file)
            else:
                print(f"❌ Failed to get connection URI for {version}")
        else:
            print(f"❌ Failed to retrieve branch ID for {version}")

    return connection_strings

def initialize_database_schema(connection_uri, schema_file):
    """Reads SQL schema from file and applies it to the database."""
    try:
        # Read schema file
        with open(schema_file, "r") as file:
            schema_sql = file.read()

        # Split SQL statements by semicolon
        sql_statements = [
            stmt.strip() for stmt in schema_sql.split(";") if stmt.strip()
        ]

        # Connect to the database
        conn = psycopg2.connect(connection_uri)
        cursor = conn.cursor()

        # Execute SQL statements one by one
        for statement in sql_statements:
            print(f"📌 Executing SQL: {statement}")
            cursor.execute(statement)

        conn.commit()
        cursor.close()
        conn.close()
        print("✅ Database schema initialized successfully.")

    except Exception as e:
        print(f"❌ Error initializing database schema: {e}")
        raise

Connecting API Version to the Correct Database Branch

We also need to retrieve the correct CONNECTION_STRINGS for database branches. To do so, we can add a helper db_connection.py Python script that returns the connection string based on the given API version:

import psycopg2
from dotenv import load_dotenv
from app.neon_db_setup import initialize_connection_strings

load_dotenv()

# Store connection strings
CONNECTION_STRINGS = initialize_connection_strings()

def get_db_connection(version: str):
    """Returns a database connection based on the API version."""
    if version not in CONNECTION_STRINGS:
        raise Exception(f"No cached connection string found for version {version}")

    return psycopg2.connect(CONNECTION_STRINGS[version])

Create FastAPI Service

Finally, we create two API routes for V1 and V2 versions. Each API version fetches different columns based on its database schema.

from fastapi import FastAPI
from app.db_connection import get_db_connection

app = FastAPI()

@app.get("/v1/users")
def get_users_v1():
    """Fetch users from v1 schema (Basic users table)."""
    conn = get_db_connection("v1")
    cur = conn.cursor()
    cur.execute("SELECT id, name, email FROM users;")
    users = cur.fetchall()
    conn.close()
    return [{"id": u[0], "name": u[1], "email": u[2]} for u in users]

@app.get("/v2/users")
def get_users_v2():
    """Fetch users from v2 schema (With additional columns)."""
    conn = get_db_connection("v2")
    cur = conn.cursor()
    cur.execute("SELECT id, name, email, age, city FROM users;")
    users = cur.fetchall()
    conn.close()
    return [
        {"id": u[0], "name": u[1], "email": u[2], "age": u[3], "city": u[4]}
        for u in users
    ]

Running the API locally

Start the API server:

uvicorn app.main:app --reload

Test Created Branches

You can easily verify branches created for API versions in the Neon Console:

Test the endpoints

Fetch users from v1 (old version)

curl -X GET "http://localhost:8000/v1/users"

Response:

[
    {"id": 1, "name": "Alice", "email": "[e-mail]"},
    {"id": 2, "name": "Bob", "email": "[e-mail]"}
]

Fetch users from v2 (new version with extra fields)

curl -X GET "http://localhost:8000/v2/users"

Response:

[
    {"id": 1, "name": "Alice", "email": "[e-mail]", "age": 25, "city": "New York"},
    {"id": 2, "name": "Bob", "email": "[e-mail]", "age": 30, "city": "San Francisco"}
]

Well done! Everything is working as its expected.

Next Steps

• You can use Neon's Schema Diff feature to track and compare schema changes between branches. The Schema Diff tool allows you to easily identify differences between the schemas of two Neon branches. This can be achieved via Neon Console or API endpoint.

• When the old API deprecates, you can make the V2 branch as a default and remove the old V1 branch safely.

Conclusion

In this project, we demonstrated how to dynamically manage API versions and database schemas using Neon database branching in FastAPI. In the next articles, we will learn how to deploy FastAPI service to Azure Cloud and use Azure API Management to route requests to the correct API version. Try out the GitHub repository example!