I have set up 2 databases in my project. One is primary which I would use for day to day use and the other is just for testing. The problem is that my Strawberry fields all use primary database and I don't know how can I properly replace that primary db with testing db.
I know that in FastAPI there's something like dependency injection with the help of which I can replace primary database connection dependency with testing database connection dependency but I don't seem to find similar solution in Strawberry.
I am using sqlalchemy with 2 postgres databases(primary and for testing), alembic for migrations and pytest with requests to perform tests.
What I'm trying to achieve:
So requests library is more suitable for an end to end testing and it's much harder to do that with a separate database, so I replaced that part with
strawberry.Schema(query=Query, mutation=Mutation).execute_sync(query="...", context_value=CustomContext())
And here's more code of what I did to accomplish what I wanted:
I replaced hard coded SessionLocal() in my endpoints with get_db() function which is being passed with a context to the endpoint:
# app/db/database.py
... # Some boiler plate code to work with database from FastAPI documentation
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# app/graphql/context.py
from strawberry.fastapi import BaseContext
from app.db.database import get_db
class Context(BaseContext):
... # Some of mine additional context code in here
def get_db(self):
return get_db()
def get_context() -> Context:
return Context()
# app/graphql/inputs.py
from app.graphql.context import Context
from strawberry.types.info import RootValueType
Info = _Info[Context, RootValueType]
# app/graphql/controllers.py
from app.graphql.inputs import Info
class Meeting:
def get_profiles_within(distance: int, info: Info):
sess = next(info.context.get_db()) # this is how I now access the db session
... # The rest of my controller code
# app/main.py
import strawberry
from strawberry.fastapi import GraphQLRouter
from app.graphql.core import Query, Mutation
from app.graphql.context import get_context
schema = strawberry.Schema(query=Query, mutation=Mutation)
graphql_app = GraphQLRouter(schema, context_getter=get_context) # This is where previously defined get_context function is used
Then I did pretty much the exact same thing for my second database. I created get_test_db() function for it and CustomContext class for it(It would've been better if I'd named it TestContext, but because this class of mine is located in the exact same file as my testing code I can't do this, because pytest automatically detects it as test):
# app/tests/db.py
... # Boiler plate code from FastAPI documentation but this time SQLALCHEMY_DATABASE_URL stores link to my test database
def get_test_db():
db = TestSessionLocal()
try:
yield db
finally:
db.close()
# app/tests/test_meeting.py
from app.tests import db
from strawberry.fastapi import BaseContext
class CustomContext(BaseContext):
... # The rest of my context, which is exactly the some code as in my regular Context class
def get_db(self):
return db.get_test_db()
And for the last part to perform mocking of the database connection I just needed to replace context when I query graphql:
# app/tests/test_meeting.py
from app.main import schema
class TestMeeting():
def test_seeking_people_around(self):
... # other testing code
response = schema.execute_sync(query=test_query, context_value=CustomContext())
by default schema doesn't have any context_value to it so I need to provide it this way for my code to work and that's exactly what I need.
Hope this helps someone who struggles setting up separate database for testing!