اتصال به دیتابیس در FastAPI

اتصال به دیتابیس PostgreSQL

برای اتصال به دیتابیس PostgreSQL ابتدا باید کتابخانه psycopg2 را نصب کنیم. برای نصب این کتابخانه از دستور زیر استفاده می‌کنیم:

pip install psycopg2

سپس برای اتصال به دیتابیس PostgreSQL و تست اتصال از کد زیر استفاده می‌کنیم:

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_posgresql_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_posgresql_connection():
        return {"message": "Successfully connected to PostgreSQL"}
    return {"message": "Failed to connection to PostgreSQL"}

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

اتصال به دیتابیس elasticsearch

برای اتصال به دیتابیس elasticsearch ابتدا باید کتابخانه elasticsearch را نصب کنیم. برای نصب این کتابخانه از دستور زیر استفاده می‌کنیم:

pip install elasticsearch

سپس برای اتصال به دیتابیس elasticsearch و تست اتصال از کد زیر استفاده می‌کنیم:

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 connection 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 connection to ElasticSearch"}

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

نکته

برای اتصال به دیتابیس‌های دیگر نیز می‌توانید از کتابخانه‌های مربوطه استفاده کنید. برای نصب کتابخانه‌های مربوط به دیتابیس‌های دیگر از دستورات زیر استفاده کنید:

#  MySQL
pip install mysql-connector-python

#  Redis
pip install redis

#  MongoDB
pip install pymongo