How to Set Up a REST API for Your Data Warehouse

Data becomes more valuable the more you can do with it. If you want to maximize the potential of your data warehouse, it helps if other systems can easily access that data. Many modern systems have a REST client that allows data to be retrieved via the HTTP protocol. By making your data warehouse available via a REST API, you instantly expand its range of applications.

In this article, you’ll learn how to add a REST API to an existing data warehouse with relative ease. The article is intended for technically savvy readers. Although the topic doesn’t directly concern BI visualizations or analyses, it does contribute to the value and usability of the BI environment within your organization.

Scenario

BI landscapes vary greatly from one organization to another and utilize a wide range of technologies. To keep this article concrete and practical, we’ve chosen a simple, common scenario. The focus is on the API itself, but the principles are broadly applicable, including to environments with cloud databases, Linux servers, or managed hosting for web applications. If you’re having trouble figuring things out within your own environment, feel free to contact us.

The Sample Landscape

We operate on the following principle:

• 💻 Platform: Microsoft SQL Server on a Windows machine

• 🔐 Network: The machine is located on an internal network with no direct access from the Internet

• 🗃️ Data warehouse: Contains fact and dimension tables with business data

• 🎯 Goal System: A customer portal that requires specific information

• 🔧 API Purpose: Provide access only to data relevant to this customer portal

• 🔐 Benefit: Granular authorization and a clear separation between technical and functional logic

So, instead of a single generic API for the entire data warehouse, we're building a targeted API for a specific application.

2. Create a virtual environment

It is good practice to always run Python solutions in their own isolated virtual environment. This simplifies package management and ensures that the solution can be moved to another machine later.

For the API files and the virtual Python environment, create a directory in a suitable location, for example C:\DWH_API_Customer Portal.

Open a PowerShell window and navigate to that directory. Create a virtual Python environment there.

PS C:\DWH_API_ClientPortal> python -m venv .venv

Next, activate this environment.

PS C:\DWH_API_CustomerPortal> .\.venv\Scripts\Activate.ps1

You can now see (.venv) will appear immediately after your prompt. This means that your isolated Python environment is active.

3. Install the required packages

Install FastAPI and the other libraries you need to run an API and connect to SQL Server:

(.venv) PS C:\DWH_API_ClientPortal> pip install fastapi uvicorn pyodbc starlette

These libraries are:

Package	Position
fastapi	Web framework for building the API
uvicorn	ASGI server to run the API
pyodbc	Connection Layer to SQL Server via ODBC
starlet	Underlying middleware for FastAPI

Your Python environment is now ready to use!

🚀Your first API endpoint

Now that your Python environment is set up, you can start writing your first API. You define FastAPI in a Python file. It's best to edit this file in a suitable editor such as Visual Studio Code (VS Code), so you can take advantage of useful features such as syntax highlighting, autocompletion, and debugging. You can download VS Code here.

1. Open VS Code

If you code if you have it available as a command-line program (via your %PATH%), you can launch the editor from the terminal:

(.venv) PS C:\DWH_API_CustomerPortal> code . -g main.py

• This will open the folder in VS Code and create a new file main.py created.
• If this doesn't work, launch VS Code by going to Start > Visual Studio Code, manually open the correct folder, and create a file yourself with the name main.py

2. Paste the following script main.py

import pyodbc, os
from fastapi import FastAPI, HTTPException, Query
from typing import Optional, List, Dict, Any

app = FastAPI()

# Adjust this connection string to fit your situation:
conn_str = (
    "DRIVER={ODBC Driver 17 for SQL Server};"
    "SERVER=localhost;"
    "DATABASE=WideWorldImportersDW;"
    "Trusted_Connection=yes;"
)

@app.get("/items")
def get_items():
    conn = pyodbc.connect(conn_str)
    cursor = conn.cursor()
    cursor.execute("SELECT ItemID, Name FROM [dwh].[Item]")
    rows = cursor.fetchall()
    return [ {"id": r.ItemID, "name": r.Name } for r in rows ]

🔍 What does this script do?

• Defines a GET-endpoint at /items
• Connects to SQL Server via ODBC
• Performs a simple SELECT on the table [dwh].[Item]
• Returns the results as a JSON list of objects containing id and name

3. Start the API

Go back to your terminal and start the API with:

(.venv) PS C:\DWH_API_ClientPortal> uvicorn main:app --reload --host 0.0.0.0 --port 8000

📌 What does this mean?

• main:app refers to the app variable in main.py
• --reload ensures that changes are applied automatically (useful during development)
• --host 0.0.0.0 makes the API accessible from other devices on the network
• --port 8000 sets the port on which the API is accessible

You should now see this message:.

INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to exit)

4. Test the endpoint

Open a browser and go to:

http://localhost:8000/items 

If everything went well, you'll see a JSON result like this:

[
 {
 "id": 0,
 "name": "Unknown"
 },
 {
 "id": 1,
 "name": "Void Fill 400 L Bag (White) 400 L"
 },
 {
 "id": 2,
 "name": "Void Fill 300 L Bag (White) 300 L"
 },
 etc. ...

Congratulations, your first API endpoint is working! 🎉

🛠️ Note: Make sure that the SQL user you're using to run the API has access to the table in question, and that the correct ODBC driver is installed (e.g., version 17 or 18 from Microsoft).

🧱 A maintainable API

The first /items-An endpoint is a good place to start, but as you add more endpoints or process larger volumes of data, it’s a good idea to structure your code more effectively. This makes maintenance easier and allows you to reuse logic.

✨ What are we improving?

• ✅ We're moving the query logic into separate functions

• ✅ We're adding error handling

• ✅ We implement page numbering for larger datasets (e.g.,. /orders)

• ✅ For bulk results, we also provide the total number of rows back

🔧 Step 1: Reusable Functions

Add at the top of your main.py Add the following functions above the endpoint definitions:

def query_sql(sql: str):
    try:
 with pyodbc.connect(conn_str) as conn:
 with conn.cursor() as cursor:
 cursor.execute(sql)
                columns = [column[0] for column in cursor.description]
 results = [dict(zip(columns, row)) for row in cursor.fetchall()]
 return results
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

def get_total_count(table: str) -> int:
    try:
    with pyodbc.connect(conn_str) as conn:
    with conn.cursor() as cursor:
                cursor.execute(f"SELECT COUNT(*) FROM {table}")
 return cursor.fetchone()[0]
    except Exception as e:
 raise HTTPException( status_code = 500, detail = str(e) )

These features:

• Connecting to SQL Server
• Execute the query
• Map the results to a list of dictionaries (JSON-compatible)
• They handle errors gracefully and provide a HTTP 500 back

🧩 Step 2: Rewrite the Items endpoint

Use the new feature in a more compact, cleaner version of /items:

@app.get( "/items", response_model = List[Dict] )
def get_items():
    sql = "SELECT * FROM dwh.Item"
    return query_sql( sql )

📄 Step 3: Orders with pagination

Pagination is essential for larger datasets. This prevents clients from having to process thousands of rows at once. Add this endpoint:

@app.get( "/orders", response_model = Dict )
def get_orders( skip: int = Query( 0, ge = 0 ), limit: int = Query( 100, ge = 1, le = 1000 ) ):
    total = get_total_count("dwh.Orders")
    sql = f"""
        SELECT * FROM dwh.Orders
 ORDER BY [Order Date Key] DESC
 OFFSET {skip} ROWS
 FETCH NEXT {limit} ROWS ONLY
    """
    items = query_sql( sql )
    return {
 "total": total,
 "skip": skip,
 "limit": limit,
 "items": items
    }

With this endpoint:

• Can you process up to 1,000 orders at a time?
• Do you donate via skip and limit which “page” section you're querying
• In addition to the results, do you also provide the total number of available rows back

Example:

http://localhost:8000/orders?limit=1000&skip=3000

📝 Best Practices

• It's better to implement complex filtering or business logic in SQL views rather than in Python code. This keeps the API generic and the logic manageable within the data warehouse.
• Make endpoints simple and predictable—with clear parameters, consistent names, and logical response structures.

🔎Filtering

A common functional requirement is that an application should only see data that is relevant to it. In this case: a customer portal that should only be able to retrieve orders from one specific customer. To do this, add a new endpoint with a path parameter instead of a query parameter.

📥 Endpoint: Orders per Customer

Add the following to main.py:

@app.get( "/customer/{customer_id}/orders", response_model = List[Dict] )
def get_orders_by_customer( customer_id: int ):
    sql = f"SELECT * FROM dwh.Orders WHERE [Customer Key] = {customer_id}"
    return query_sql( sql )

• customer_id is part of the URL itself (a so-called path parameter).
• This value is used directly in the SQL query.

Example:

http://localhost:8000/customer/50/orders.

…retrieves all orders for the customer with ID 50.

⚠️ Safety Tip

The use of f"SELECT ... {value}" is vulnerable to SQL injection if you don't have control over the input. For added security (especially with text fields), use parameter binding with ?

1. Edit your job title query_sql so that it accepts optional parameters:

from typing import Optional, List, Any

def query_sql(sql: str, params: Optional[List[Any]] = None):
    try:
 with pyodbc.connect(conn_str) as conn:
            with conn.cursor() as cursor:
 if params:
 cursor.execute(sql, params)
 else:
 cursor.execute(sql)
 columns = [column[0] for column in cursor.description]
                results = [dict(zip(columns, row)) for row in cursor.fetchall()]
 return results
    except Exception as e:
 raise HTTPException(status_code=500, detail=str(e))

2. In your endpoint, you now send the SQL query with placeholders (?) along with a list of parameters:

@app.get("/customer/{customer_id}/orders", response_model=List[Dict])
def get_orders_by_customer(customer_id: int):
    sql = "SELECT * FROM dwh.Orders WHERE [Customer Key] = ?"
    return query_sql(sql, [customer_id])

Why does this work?

• PyODBC (like most database APIs in Python) supports parameter binding via ? placeholders in the SQL string.
• Parameters are sent separately from the query, making SQL injection impossible.
• By modifying the generic function with an optional params parameter, you can still use the same function for simple queries without parameters.

🧼 Additional filter: open orders only

Suppose you want to retrieve only open orders. In that case, you can create a second filtered endpoint:

@app.get("/customer/{customer_id}/open_orders", response_model=List[Dict])
def get_open_orders_by_customer(customer_id: int):
    sql = f"""
        SELECT * FROM dwh.Orders
        WHERE [Customer Key] = {customer_id} AND [Order Status] = 'open'
    """
    return query_sql(sql)

🎯 Best Practices

Define a view in the data warehouse that contains only the open orders per customer, and call it from the API. This way, all the logic is managed in one central location.

📝 Note the data type of customer numbers 

The examples above assume that the customer number is a int is. In your own data warehouse, that can also be a varchar, guid, or something else. Only then should you change the type designation customer_id: int to the correct type, for example str.

⚙️Generate endpoints

If your data warehouse contains many tables and you don't want to manually write an endpoint for each table, you can automate this process. By running a smart SQL query on the SQL Server system catalog, you can generate Python code that you can paste directly into your API.

🎯 Goal

• Quickly create API endpoints for all tables in certain diagrams (for example, Fact and Dimension)
• Support for endpoints that search for primary key (detail endpoints)
• Endpoints with pagination for bulk data

📋 Query 1: Endpoints by primary key

This query generates a FastAPI endpoint for each table, which retrieves a record based on the primary key (assuming a single column as the primary key).

SELECT '@app.get( "/' + REPLACE( KCU.TABLE_NAME, ' ', '' ) + '/{' + REPLACE( KCU.TABLE_NAME, ' ', '' ) + '_id}", response_model = List[Dict] )
def get_' + REPLACE( KCU.TABLE_NAME, ' ', '' ) + '_by_id( ' + REPLACE( KCU.TABLE_NAME, ' ', '' ) + '_id: ' + case C.DATA_TYPE WHEN 'int' then 'int' WHEN 'bigint' then 'int' WHEN 'varchar' then 'str' WHEN 'date' then 'str' END + ' ):
    sql = f"SELECT * FROM [' + KCU.TABLE_CATALOG + '].[' + KCU.TABLE_SCHEMA + '].[' + KCU.TABLE_NAME + '] WHERE [' + KCU.COLUMN_NAME + '] = {' + REPLACE( KCU.TABLE_NAME, ' ', '' ) + '_id}"
    return query_sql( sql )

'
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE KCU
JOIN INFORMATION_SCHEMA.TABLE_CONSTRAINTS TC ON KCU.CONSTRAINT_NAME = TC.[CONSTRAINT_NAME] AND KCU.CONSTRAINT_SCHEMA = TC.[CONSTRAINT_SCHEMA] AND TC.[CONSTRAINT_TYPE] = 'PRIMARY KEY'
JOIN INFORMATION_SCHEMA.COLUMNS C ON KCU.TABLE_CATALOG = C.TABLE_CATALOG AND KCU.TABLE_SCHEMA = C.TABLE_SCHEMA AND KCU.TABLE_NAME = C.TABLE_NAME AND KCU.COLUMN_NAME = C.COLUMN_NAME
WHERE KCU.TABLE_SCHEMA IN ('Fact', 'Dimension') AND KCU.ORDINAL_POSITION = 1

 

📋 Query 2: Bulk endpoints with pagination

This query generates endpoints that paginate data by table, which is useful for tables with many records.

SELECT '@app.get( "/all_' + REPLACE( KCU.TABLE_NAME, ' ', '' ) + '", response_model = Dict )
def get_all_' + REPLACE( KCU.TABLE_NAME, ' ', '' ) + '( skip: int = Query( 0, ge = 0 ), limit: int = Query( 100, ge = 1, le = 1000 ) ):
    total = get_total_count("[' + KCU.TABLE_CATALOG + '].[' + KCU.TABLE_SCHEMA + '].[' + KCU.TABLE_NAME + ']")
    sql = f"""
        SELECT * FROM [' + KCU.TABLE_CATALOG + '].[' + KCU.TABLE_SCHEMA + '].[' + KCU.TABLE_NAME  + ']
		ORDER BY [' +  KCU.COLUMN_NAME + ']
 OFFSET {skip} ROWS
 FETCH NEXT {limit} ROWS ONLY
    """
    items = query_sql( sql )
    return { "total": total, "skip": skip, "limit": limit, "values": items }

'
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE KCU
JOIN INFORMATION_SCHEMA.TABLE_CONSTRAINTS TC ON KCU.CONSTRAINT_NAME = TC.[CONSTRAINT_NAME] AND KCU.CONSTRAINT_SCHEMA = TC.[CONSTRAINT_SCHEMA] AND TC.[CONSTRAINT_TYPE] = 'PRIMARY KEY'
JOIN INFORMATION_SCHEMA.COLUMNS C ON KCU.TABLE_CATALOG = C.TABLE_CATALOG AND KCU.TABLE_SCHEMA = C.TABLE_SCHEMA AND KCU.TABLE_NAME = C.TABLE_NAME AND KCU.COLUMN_NAME = C.COLUMN_NAME
WHERE KCU.TABLE_SCHEMA IN ('Fact','Dimension') AND KCU.ORDINAL_POSITION = 1

⚠️ Please note

• These queries assume that all tables have a a single column as the primary key have (no composite keys).
• If necessary, adjust the schema names (Fact, Dimension) to your data warehouse environment.
• Paste the output of these queries into your main.py below your existing code.

💡 Tip

Do you want to combine the generated endpoints with the parameter-bound secure query_sql() function, then modify the generated SQL strings using ? and include parameters, as discussed earlier.

🧬 Good to Know: Data Type Conversion Between SQL and JSON

When you expose data from your SQL database to JSON via FastAPI, SQL data types are automatically converted to JSON-compatible formats. This happens because the database driver (such as pyodbc) converts the SQL results to Python types, after which FastAPI serializes these Python types to JSON.

The most common SQL data types, such as integers, strings, and booleans, are converted directly to JSON numbers, strings, and booleans, respectively. Dates and times are automatically converted to ISO 8601-formatted strings, making them readable and standardized in JSON. Null values in the database appear as null in JSON.

For more complex or specialized data types, such as binary data or geographic types, you may need to implement your own conversions or validations. This allows you to maintain control over how data appears in JSON and ensures consistent and accurate API responses.

SQL data type	Python data type	JSON data type	Note
int, bigint	int	Number (integer)	Converted immediately
float, decimal	float / Decimal	Number (float)	Decimal is usually converted to float
varchar, text	str	String	Converted immediately
bit	bool	Boolean	Will be true/false
datetime, date	datetime.datetime / datetime.date	String (ISO 8601)	Becomes an ISO 8601 string (e.g.,. "2025-07-02T10:30:00")
time	datetime.time	String (HH:MM:SS)	Becomes a thong
NULL	None	null	JSON null

🔐A (more) secure API with an API key

Without security measures in place, anyone with network access can use your API. That's usually not desirable, even within an internal network. That's why we're adding a simple, centralized API key authentication.

🛠️ Preparation: Set up an API key

1. Generate a long, strong API key.

For example, using an online generator or PowerShell:

2. Set this API key as an environment variable on the machine where the API is running, for example:

(.venv) PS C:\DWH_API_CustomerPortal> $env:API_KEY = "your-long-api-key-here"

The command above makes the variable available in the current PowerShell session. Also, make sure it is a global system variable so that it is preserved after a reboot.

🧩 Code changes

Add at the top of your main.py Add the following lines:

import os

API_KEY = os.getenv("DWH_API_KEY")
if not API_KEY:
    raise RuntimeError("The DWH_API_KEY environment variable was not found. Make sure it is set.")

🚧 Add middleware for API key validation

Implement Starlette middleware that checks every incoming request for the correct API key in the header X-API-Key:

from fastapi.responses import Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request

class APIKeyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
    if request.headers.get("X-API-Key") != API_KEY:
            return Response(status_code = 403, content = "Forbidden")
return await call_next(request)

app = FastAPI()
app.add_middleware( APIKeyMiddleware )

🔑 Using the API

• Required: Include the following header with each request: X-API-Key Tighten it with the correct wrench.
• For testing, you can use Postman, browsers such as Firefox with developer tools, or command-line tools such as curl:

curl -H "X-API-Key: your-long-api-key-here" http://localhost:8000/items

⚙️To Production

Run the API as a Windows service

If your API runs on a Windows machine, you'll want it to start automatically when the server boots up and to keep running at all times, even after crashes or restarts. The easiest way to do this is to install the API as a Windows service.

🛠️ What do you need?

• nssm (Non-Sucking Service Manager), a handy little tool for running any script as a Windows service.
• A dedicated, non-privileged Windows user for the service (for security and file permissions).
• Access to the folder containing your Python virtual environment and API code.
• The appropriate permissions to install and manage services (admin permissions).

📝 Example: batch script for nssm

Create a .bat-file, for example start_dwh_api.bat with the following content:

@echo off
cd /d "C:\DWH_API_CustomerPortal"
call .venv\Scripts\activate.bat
python -m uvicorn main:app --host 0.0.0.0 --port 8000

🔧 Installing a service with nssm

1. Download nssm from https://nssm.cc/download
2. Open a command prompt as an administrator
3. Install the service using nssm:

PS C:\> nssm install DWH_API_Service "C:\Path\to\start_dwh_api.bat"

4. Configure the service in nssm (using the GUI or command line) to run under the correct user and with the correct working directory.
5. Start the service

PS C:\> nssm start DWH_API_Service

🔐 Rights and Safety

• Give the Windows user only SELECT permissions on the required database views.
• Give this user read and execute permissions in the folder containing the API and the virtual environment.
• Avoid running the service under an Administrator account.

🌐 API Availability

Your API is now up and running, but how do you ensure that other systems can access it? This depends on where your API server is located and where the consuming applications are running.

🖥️ Local access

• If the API is running on a machine within the same (virtual) network as the consuming application, you can use that machine's IP address or hostname instead of localhost.
• For example:

http://192.168.1.100:8000/items

or

http://api-server-01.yourdomain.local:8000/items

🌍 Remote access via the Internet

If applications need to access the API via the Internet:

• Make sure to Fully Qualified Domain Name (FQDN) which refers to the public IP address of your API host (or an intermediate firewall/load balancer).
• Configure DNS so that the FQDN points to that IP address.
• Set a reverse proxy or load balancer between the Internet and your API host, such as nginx, Azure Application Gateway, or AWS ALB.
• This intermediate layer often also handles TLS (HTTPS) termination and routing.

🔒 Network-layer security

• Always use TLS (HTTPS) to encrypt traffic.
• Configure your firewall so that only traffic from specific IP addresses or networks is allowed.
• Consider using a VPN to access internal applications.

🔧 FastAPI Configuration for HTTPS Redirect

FastAPI itself cannot handle TLS, but you can configure FastAPI to automatically redirect HTTP requests to HTTPS:

from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()
app.add_middleware( HTTPSRedirectMiddleware )

Please note: Your reverse proxy or load balancer must handle TLS.

💡 Summary

Scenario	Example of an access address	Comments
Same network	http://192.168.1.100:8,000/items	Easy to access via IP or hostname
Public Internet	https://api.jouwbedrijf.nl/items	Use DNS, reverse proxy, TLS
Development (local)	http://localhost:8000/items	Available only locally

📚 You’ve got the docs! — Automatic API documentation

FastAPI comes with a major bonus: automatic, interactive documentation. This makes it easy to explore and test your API, and it helps you communicate with other teams or external developers.

🔎 What's included?

• Swagger UI:
  A web interface where you can view all available endpoints, parameters, and response models and try them out right away.
  Visit it at:

http://localhost:8000/docs

• ReDoc:
  An alternative, more technical API documentation that is well-organized and easy to read.
  Available at:

http://localhost:8000/redoc

• OpenAPI spec:
  The complete machine-readable API specification in JSON format, which allows other tools and systems to understand your API:

http://localhost:8000/openapi.json

🎯 Why is this useful?

• Faster testing and development: without having to write documentation yourself or use separate tools.
• Communication: Easily share with developers, support, or external parties.
• Automation: The tooling can automatically generate client libraries based on the OpenAPI specification.
• Up to date: The documentation is always in sync with your code because it's generated in real time.

💡 Tips

• Usage type hints and response_model in your endpoints to ensure the documentation is complete and accurate.
• Add descriptions where necessary using FastAPI’s Query(), Path(), and Body() parameters.
• For secure APIs, make sure your testing tools also include the required API key header so you can test the endpoints.

🌐 API Standards: FastAPI and OpenAPI vs. Others

FastAPI is fully compliant with OpenAPI (formerly Swagger), which provides clear, machine-readable documentation and broad support from various tools. This makes it easy to design and document RESTful APIs.

However, FastAPI does not natively support other API standards such as OData, which is best known for its extensive query and filtering capabilities within URLs and is widely used in Microsoft environments.

Other well-known API standards include:

• GraphQL: enables clients to precisely define what data they want to receive; it is often used for complex front-end applications.
• gRPC: A protocol for fast, type-safe communication between microservices.
• JSON:API: a standard for consistent structure and communication in RESTful APIs.
• SOAP: An older, XML-based web service standard that is still commonly used in many enterprise environments.

Although FastAPI focuses on modern RESTful APIs with OpenAPI, it is possible to support some other standards using additional libraries, but that requires extra configuration and customization. FastAPI is a top choice for RESTful APIs using HTTP/JSON and OpenAPI documentation. Other standards require different tools and approaches that FastAPI does not support out-of-the-box.

🎯 Conclusion

With a relatively simple FastAPI implementation, you can make your data warehouse much more accessible to modern applications. By using the right structure, security measures, and standard tools, you can quickly build a reliable REST API that adds immediate value.

The result:

• Greater ease of use: Other systems can retrieve data in a standardized, web-friendly manner.
• Improved manageability: Business logic stays where it belongs—in the data warehouse—and APIs remain simple and maintainable.
• Safety: Using API keys and HTTPS prevents unauthorized access and protects your data in transit.
• Scalability: By paginating and generating endpoints, you can also keep larger datasets manageable.

Do you have any questions or need help building your API? Feel free to contact us!

📢 Disclaimer

This article was created in collaboration with an AI language model. The content is intended to provide you with the best possible support, but always maintain a critical mindset and adapt the examples to your specific situation and security requirements.