Database
Starlette is not strictly tied to any particular database implementation.
You can use it with an asynchronous ORM, such asGINO,or use regular non-async endpoints, and integrate withSQLAlchemy.
In this documentation we'll demonstrate how to integrate againstthedatabases package,which provides SQLAlchemy core support against a range of different database drivers.
Here's a complete example, that includes table definitions, configuring adatabase.Databaseinstance, and a couple of endpoints that interact with the database.
DATABASE_URL=sqlite:///test.dbimportcontextlibimportdatabasesimportsqlalchemyfromstarlette.applicationsimportStarlettefromstarlette.configimportConfigfromstarlette.responsesimportJSONResponsefromstarlette.routingimportRoute# Configuration from environment variables or '.env' file.config=Config('.env')DATABASE_URL=config('DATABASE_URL')# Database table definitions.metadata=sqlalchemy.MetaData()notes=sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)database=databases.Database(DATABASE_URL)@contextlib.asynccontextmanagerasyncdeflifespan(app):awaitdatabase.connect()yieldawaitdatabase.disconnect()# Main application code.asyncdeflist_notes(request):query=notes.select()results=awaitdatabase.fetch_all(query)content=[{"text":result["text"],"completed":result["completed"]}forresultinresults]returnJSONResponse(content)asyncdefadd_note(request):data=awaitrequest.json()query=notes.insert().values(text=data["text"],completed=data["completed"])awaitdatabase.execute(query)returnJSONResponse({"text":data["text"],"completed":data["completed"]})routes=[Route("/notes",endpoint=list_notes,methods=["GET"]),Route("/notes",endpoint=add_note,methods=["POST"]),]app=Starlette(routes=routes,lifespan=lifespan,)Finally, you will need to create the database tables. It is recommended to useAlembic, which we briefly go over inMigrations
Queries
Queries may be made with asSQLAlchemy Core queries.
The following methods are supported:
rows = await database.fetch_all(query)row = await database.fetch_one(query)async for row in database.iterate(query)await database.execute(query)await database.execute_many(query)
Transactions
Database transactions are available either as a decorator, as acontext manager, or as a low-level API.
Using a decorator on an endpoint:
@database.transaction()asyncdefpopulate_note(request):# This database insert occurs within a transaction.# It will be rolled back by the `RuntimeError`.query=notes.insert().values(text="you won't see me",completed=True)awaitdatabase.execute(query)raiseRuntimeError()Using a context manager:
asyncdefpopulate_note(request):asyncwithdatabase.transaction():# This database insert occurs within a transaction.# It will be rolled back by the `RuntimeError`.query=notes.insert().values(text="you won't see me",completed=True)awaitrequest.database.execute(query)raiseRuntimeError()Using the low-level API:
asyncdefpopulate_note(request):transaction=awaitdatabase.transaction()try:# This database insert occurs within a transaction.# It will be rolled back by the `RuntimeError`.query=notes.insert().values(text="you won't see me",completed=True)awaitdatabase.execute(query)raiseRuntimeError()except:awaittransaction.rollback()raiseelse:awaittransaction.commit()Test isolation
There are a few things that we want to ensure when running tests againsta service that uses a database. Our requirements should be:
- Use a separate database for testing.
- Create a new test database every time we run the tests.
- Ensure that the database state is isolated between each test case.
Here's how we need to structure our application and tests in order tomeet those requirements:
fromstarlette.applicationsimportStarlettefromstarlette.configimportConfigimportdatabasesconfig=Config(".env")TESTING=config('TESTING',cast=bool,default=False)DATABASE_URL=config('DATABASE_URL',cast=databases.DatabaseURL)TEST_DATABASE_URL=DATABASE_URL.replace(database='test_'+DATABASE_URL.database)# Use 'force_rollback' during testing, to ensure we do not persist database changes# between each test case.ifTESTING:database=databases.Database(TEST_DATABASE_URL,force_rollback=True)else:database=databases.Database(DATABASE_URL)We still need to setTESTING during a test run, and setup the test database.Assuming we're usingpy.test, here's how ourconftest.py might look:
importpytestfromstarlette.configimportenvironfromstarlette.testclientimportTestClientfromsqlalchemyimportcreate_enginefromsqlalchemy_utilsimportdatabase_exists,create_database,drop_database# This sets `os.environ`, but provides some additional protection.# If we placed it below the application import, it would raise an error# informing us that 'TESTING' had already been read from the environment.environ['TESTING']='True'importapp@pytest.fixture(scope="session",autouse=True)defcreate_test_database():""" Create a clean database on every test case. For safety, we should abort if a database already exists. We use the `sqlalchemy_utils` package here for a few helpers in consistently creating and dropping the database. """url=str(app.TEST_DATABASE_URL)engine=create_engine(url)assertnotdatabase_exists(url),'Test database already exists. Aborting tests.'create_database(url)# Create the test database.metadata.create_all(engine)# Create the tables.yield# Run the tests.drop_database(url)# Drop the test database.@pytest.fixture()defclient():""" When using the 'client' fixture in test cases, we'll get full database rollbacks between test cases: def test_homepage(client): url = app.url_path_for('homepage') response = client.get(url) assert response.status_code == 200 """withTestClient(app)asclient:yieldclientMigrations
You'll almost certainly need to be using database migrations in order to manageincremental changes to the database. For this we'd strongly recommendAlembic, which is written by the author of SQLAlchemy.
$pipinstallalembic$alembicinitmigrationsNow, you'll want to set things up so that Alembic references the configuredDATABASE_URL, and uses your table metadata.
Inalembic.ini remove the following line:
sqlalchemy.url=driver://user:pass@localhost/dbnameInmigrations/env.py, you need to set the'sqlalchemy.url' configuration key,and thetarget_metadata variable. You'll want something like this:
# The Alembic Config object.config=context.config# Configure Alembic to use our DATABASE_URL and our table definitions...importappconfig.set_main_option('sqlalchemy.url',str(app.DATABASE_URL))target_metadata=app.metadata...Then, using our notes example above, create an initial revision:
alembicrevision-m"Create notes table"And populate the new file (withinmigrations/versions) with the necessary directives:
defupgrade():op.create_table('notes',sqlalchemy.Column("id",sqlalchemy.Integer,primary_key=True),sqlalchemy.Column("text",sqlalchemy.String),sqlalchemy.Column("completed",sqlalchemy.Boolean),)defdowngrade():op.drop_table('notes')And run your first migration. Our notes app can now run!
alembicupgradeheadRunning migrations during testing
It is good practice to ensure that your test suite runs the database migrationsevery time it creates the test database. This will help catch any issues in yourmigration scripts, and will help ensure that the tests are running againsta database that's in a consistent state with your live database.
We can adjust thecreate_test_database fixture slightly:
fromalembicimportcommandfromalembic.configimportConfigimportapp...@pytest.fixture(scope="session",autouse=True)defcreate_test_database():url=str(app.DATABASE_URL)engine=create_engine(url)assertnotdatabase_exists(url),'Test database already exists. Aborting tests.'create_database(url)# Create the test database.config=Config("alembic.ini")# Run the migrations.command.upgrade(config,"head")yield# Run the tests.drop_database(url)# Drop the test database.