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, Nov 23 '21 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, Nov 23 '21 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. https://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, Nov 23 '21 at 14:43

score 3 · Answer 1 · edited Aug 31 '22 at 15:24

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.

This solution worked for me, but it's probably good to note that you'll need to add `app.add_middleware(DatabaseSelector)` — Tunnelvisie, Jul 26 '22 at 09:53

Multiple Database connections using fastapi

1 Answers1

Linked