Connecting to a Database in FastAPI

Connecting to a PostgreSQL Database

To connect to a PostgreSQL database, you first need to install the psycopg2 library. Use the following command to install this library:

pip install psycopg2

Then, to connect to the PostgreSQL database and test the connection, use the following code:

import os
import psycopg2
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI()

# Database connection parameters
DB_USER = os.environ.get('DB_USER', '')
DB_PASSWORD = os.environ.get('DB_PASSWORD', '')
DB_HOST = os.environ.get('DB_HOST', '')
DB_PORT = os.environ.get('DB_PORT', '')
DB_NAME = os.environ.get('DB_NAME', '')

def check_postgresql_connection():
    try:
        conn = psycopg2.connect(
            user=DB_USER,
            password=DB_PASSWORD,
            host=DB_HOST,
            port=DB_PORT,
            database=DB_NAME
        )
        conn.close()
        return True
    except Exception as e:
        return False

class Item(BaseModel):
    name: str

@app.post("/")
async def home(item: Item):
    if check_postgresql_connection():
        return {"message": "Successfully connected to PostgreSQL"}
    return {"message": "Failed to connect to PostgreSQL"}

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get('PORT', 5000)))

Connecting to an Elasticsearch Database

To connect to an Elasticsearch database, you first need to install the elasticsearch library. Use the following command to install this library:

pip install elasticsearch

Then, to connect to the Elasticsearch database and test the connection, use the following code:

import os
from fastapi import FastAPI
from elasticsearch import Elasticsearch
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

ELASTIC_USER = os.environ.get('ELASTIC_USER')
ELASTIC_PASS = os.environ.get('ELASTIC_PASS')
ELASTIC_DOMAIN = os.environ.get('ELASTIC_DOMAIN')

app = FastAPI()

def check_elastic_search_connection():
    elastic_search = Elasticsearch(f'{ELASTIC_DOMAIN}', basic_auth=(ELASTIC_USER, ELASTIC_PASS))
    elastic_ping = elastic_search.ping()
    if elastic_ping:
        logging.info(f"Successfully connected to ElasticSearch: {elastic_ping}")
        return True
    else:
        logging.error(f"Failed to connect to ElasticSearch: {elastic_ping}")
        return False

@app.get("/items/")
def elastic_search_get():
    if check_elastic_search_connection():
        return {"message": "Successfully connected to ElasticSearch"}
    return {"message": "Failed to connect to ElasticSearch"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

tip

You can also use the relevant libraries to connect to other databases. Use the following commands to install the libraries for other databases:

# MySQL
pip install mysql-connector-python

# Redis
pip install redis

# MongoDB
pip install pymongo