Multiple Database connections using fastapi

Question

I am new to FAST API and don't have much knowledge on the same. I wanted to connect multiple databases. I tried to go through the https://github.com/tiangolo/fastapi/issues/2592 but was unable to understand the approach. I was also curious as to how will I be able to fetch data from the database and also update the values in the database. It will be great if anyone can help me out, also are there any other examples or methods which would do the job?

The issue you've linked shows you can create two engines and explicitly links models to each engine. What is it about the answer you don't understand? You can also have two dependencies (i.e. get_db and get_db_alternative) and yield different sessions based on what you need. — MatsLindh
– MatsLindh, Commented Nov 23, 2021 at 13:49
@MatsLindh 1) the role of as_declarative 2) how will I be able to fetch data from the database and also update the values in the database — Rahul
– Rahul, Commented Nov 23, 2021 at 13:57
as_declarative allows you to mark a class as a declarative base class in SQLAlchemy, so that you could create models that inherit from either Model1DB or Model2DB based on which DB the model lives in (instead of using Base). For querying and updating it would depend on how you either define your models or are planning to use queries or table objects in SQLAlchemy. Those are general SQLAlchemy questions; i.e. docs.sqlalchemy.org/en/14/core/dml.html is a good resource to find out how to run queries against an engine or a session. — MatsLindh
– MatsLindh, Commented Nov 23, 2021 at 14:43

Javad · Accepted Answer · 2022-08-31 15:24:48Z

3

Make yourself a middleware.Here is my example:

# extends BaseHTTPMiddleware filter your router name and set database_url in your envioment.
class DatabaseSelector(BaseHTTPMiddleware):
    def __init__(self, app):
        super().__init__(app)

    async def dispatch(self, request: Request, call_next):
        url_param = str(request.url).split("/")
        if "api" in str(request.url):
            url_param.remove("api")
        route_name = url_param[3]
        if route_name not in list(routeMap.keys()):
            SQLALCHEMY_DATABASE_URI = routeMap["base_mysql"]
        else:
            SQLALCHEMY_DATABASE_URI = routeMap[route_name]
        os.environ["SQLALCHEMY_DATABASE_URI"] = SQLALCHEMY_DATABASE_URI
        response = await call_next(request)
        return response

And make your config params like this:

routeMap = {
    "base_mysql": 'mysql+pymysql://{}:{}@{}:{}/{}'.format(*BaseMysql.values()),
    "show_data": 'mysql+pymysql://{}:{}@{}:{}/{}'.format(*RequestMysql.values()),
    "else_data": "xxxxxxx",
    }

while send http request. you can get your router name base on this routeMap:

def get_db():
    SQLALCHEMY_DATABASE_URI = os.getenv('SQLALCHEMY_DATABASE_URI')
    engine = create_engine(SQLALCHEMY_DATABASE_URI, echo=True, pool_size=50)
    SessionLocal = sessionmaker(bind=engine)
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

finally, you just call get_db without care which database you will connect.

edited Aug 31, 2022 at 15:24

Javad

2,1333 gold badges16 silver badges25 bronze badges

answered Jul 21, 2022 at 9:37

MG-He

314 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Tunnelvisie Over a year ago

This solution worked for me, but it's probably good to note that you'll need to add app.add_middleware(DatabaseSelector)

Collectives™ on Stack Overflow

Multiple Database connections using fastapi

1 Answer 1

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related