{"id":71,"date":"2024-11-10T10:00:00","date_gmt":"2024-11-10T16:00:00","guid":{"rendered":"https:\/\/douglasstarnes.dev\/?p=71"},"modified":"2024-11-09T10:49:35","modified_gmt":"2024-11-09T16:49:35","slug":"database-driven-fastapi","status":"publish","type":"post","link":"https:\/\/douglasstarnes.dev\/index.php\/2024\/11\/10\/database-driven-fastapi\/","title":{"rendered":"Database Driven FastAPI"},"content":{"rendered":"\n

In the previous post, you saw how to implement a complete REST API. To keep things simple, all of the data was stored in a Python dictionary. This is a less than ideal solution for a number of reasons. A real world application would use a database. FastAPI works with a number of different databases and can accommodate a flexible array of development scenarios. In this post, we will use a combination of SQLite, SQLAlchemy, and Alembic, in cooperation with the Pydantic schemas from the previous post to refactor the application to use a database to store the todo list.<\/p>\n\n\n\n

Application Structure<\/h2>\n\n\n\n

As this application is more complex than the previous, we should structure it so that it is easier to maintain and understand. There will be separate modules for the Pydantic schemas, the SQLAlchemy models, a repository that handles the interaction with the database, utilities to manage the database, and the FastAPI instance itself. All of these will be inside of a parent module containing the FastAPI application.<\/p>\n\n\n\n

I’ll put the parent module in a new folder called app<\/code>. Inside of app<\/code>, I’ll add a blank __init__.py<\/code> file to mark the folder as a module.<\/p>\n\n\n\n

Database Utilities<\/h2>\n\n\n\n

Before going further, now is a good time to install the required packages. (The pydantic<\/code> package is installed as a dependency of fastapi<\/code>.)<\/p>\n\n\n\n

<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
$<\/span> <\/span>pip<\/span> <\/span>install<\/span> <\/span>fastapi<\/span> <\/span>sqlalchemy<\/span> <\/span>alembic<\/span> <\/span>uvicorn<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Next, create a new file, inside of app<\/code>, named db.py<\/code>.  This will hold the database utilities.  Here is where we will configure SQLAlchemy to connect to a SQLite database.  The following code imports several functions from the sqlalchemy<\/code> module to connect to the database.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> sqlalchemy <\/span>import<\/span> create_engine<\/span><\/span>\nfrom<\/span> sqlalchemy<\/span>.<\/span>orm <\/span>import<\/span> DeclarativeBase<\/span>,<\/span> sessionmaker<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The create_engine<\/code> function will return a SQLAlchemy Engine<\/code> which is used by the sessionmaker<\/code> function to create a session factory.  The sessions will be used in the application to query the database.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
DATABASE_URL <\/span>=<\/span> <\/span>"<\/span>sqlite:\/\/\/.\/todo.db<\/span>"<\/span><\/span>\n<\/span>\nengine <\/span>=<\/span> <\/span>create_engine<\/span>(<\/span>DATABASE_URL<\/span>,<\/span> <\/span>connect_args<\/span>=<\/span>{<\/span>"<\/span>check_same_thread<\/span>"<\/span>:<\/span> <\/span>False<\/span>})<\/span><\/span>\nSession <\/span>=<\/span> <\/span>sessionmaker<\/span>(<\/span>autocommit<\/span>=False<\/span>,<\/span> <\/span>autoflush<\/span>=False<\/span>,<\/span> <\/span>bind<\/span>=<\/span>engine<\/span>)<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
As SQLite is a file-based database, the connection string is just a file path. The three slashes after the protocol name mean that this is a relative path. So relative to the current path SQLAlchemy will look for a file named blog.db<\/code>. I won’t go into great detail about the check_same_thread<\/code> argument. It’s something that is specific to the combination of SQLite and FastAPI. However, it is intended to prevent conflicts when accessing the database from different threads.<\/p>\n\n\n\n
The sessionmaker<\/code> function binds to the engine<\/code> to create sessions to interact with the database.  The autocommit<\/code> and autoflush<\/code> keyword arguments give you absolute control when persisting data to the database.  In web applications, it is best to keep control so that you ensure database operations are completed in the context of a request.  Thus these values are set to False<\/code>.<\/p>\n\n\n\n
You’ll also need a base class from which to derive the SQLAlchemy models. For this demo, you can just derive a class from DeclarativeBase.  For more complex applications you might include configuration and other setup.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
class<\/span> <\/span>Base<\/span>(<\/span>DeclarativeBase<\/span>):<\/span><\/span>\n  <\/span>pass<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Pydantic Schemas<\/h2>\n\n\n\n
We’ll come back to the db<\/code> module later.  Let’s work on the models and the schemas.  The schemas you saw in the previous post.  These are classes, derived from the Pydantic BaseModel<\/code>, that represent the structure of the data inside of the FastAPI application, as opposed to the SQLAlchemy models that represent the structure of the database.<\/p>\n\n\n\n
Inside of the app<\/code> module folder, create a new file schemas.py<\/code>, and import the BaseModel<\/code> from the pydantic<\/code> module.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> pydantic <\/span>import<\/span> BaseModel<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The implementation of the TodoItem<\/code> schema starts off the same.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
class<\/span> <\/span>TodoItem<\/span>(<\/span>BaseModel<\/span>):<\/span><\/span>\n  <\/span>id<\/span>:<\/span> <\/span>int<\/span><\/span>\n  task<\/span>:<\/span> <\/span>str<\/span><\/span>\n  complete<\/span>:<\/span> <\/span>bool<\/span><\/span>\n  <\/span><\/span>\n  <\/span>class<\/span> <\/span>Config<\/span>:<\/span><\/span>\n    orm_mode <\/span>=<\/span> <\/span>True<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The Config<\/code> inner class is used to set addition configuration options for how the schema should behave.  In this case, orm_mode<\/code> makes is easier to convert between the Pydantic schema classes and the SQLAlchemy model classes that will be created in the next section.<\/p>\n\n\n\n
But first, go ahead and add the two DTO schemas for creating and updating TodoItem<\/code>s.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> typing <\/span>import<\/span> Optional<\/span><\/span>\n<\/span>\nclass<\/span> <\/span>CreateTodoItem<\/span>(<\/span>BaseModel<\/span>):<\/span><\/span>\n  task<\/span>:<\/span> <\/span>str<\/span><\/span>\n  <\/span><\/span>\nclass<\/span> <\/span>UpdateTodoItem<\/span>(<\/span>BaseModel<\/span>):<\/span><\/span>\n  task<\/span>:<\/span> Optional<\/span>[<\/span>str<\/span>]<\/span> <\/span>=<\/span> <\/span>None<\/span><\/span>\n  complete<\/span>:<\/span> Optional<\/span>[<\/span>bool<\/span>]<\/span> <\/span>=<\/span> <\/span>None<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
As these are unchanged from the previous post I won’t discuss them further.  And that’s all for the Pydantic schema classes!<\/p>\n\n\n\n
SQLAlchemy Models<\/h2>\n\n\n\n
There is only one model class for this application.  Put it in new Python file, models.py<\/code>, in the app<\/code> module folder.<\/p>\n\n\n\n
In SQLAlchemy, a model class derives from the base class generated by the declarative_base<\/code> factory function earlier from the db<\/code> module.  In the model class, you use types from the sqlalchemy<\/code> module to define the mappings to columns in a database table.  Here are the required imports.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> sqlalchemy <\/span>import<\/span> Integer<\/span>,<\/span> String<\/span>,<\/span> Boolean<\/span><\/span>\nfrom<\/span> sqlalchemy<\/span>.<\/span>orm <\/span>import<\/span> Mapped<\/span>,<\/span> mapped_column<\/span><\/span>\n<\/span>\nfrom<\/span> <\/span>.<\/span>db <\/span>import<\/span> Base<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
In the model class, a database table column is represented with the Mapped<\/code> class which use Python type hints.  These are generated with the mapped_column<\/code> function.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
class<\/span> <\/span>TodoItem<\/span>(<\/span>Base<\/span>):<\/span><\/span>\n  __tablename__ <\/span>=<\/span> <\/span>"<\/span>todo_items<\/span>"<\/span><\/span>\n  <\/span><\/span>\n  <\/span>id<\/span>:<\/span> Mapped<\/span>[<\/span>int<\/span>]<\/span> <\/span>=<\/span> <\/span>mapped_column<\/span>(<\/span>Integer<\/span>,<\/span> <\/span>primary_key<\/span>=True<\/span>)<\/span><\/span>\n  task<\/span>:<\/span> Mapped<\/span>[<\/span>str<\/span>]<\/span> <\/span>=<\/span> <\/span>mapped_column<\/span>(<\/span>String<\/span>)<\/span><\/span>\n  complete<\/span>:<\/span> Mapped<\/span>[<\/span>bool<\/span>]<\/span> <\/span>=<\/span> <\/span>mapped_column<\/span>(<\/span>Boolean<\/span>,<\/span> <\/span>default<\/span>=False<\/span>)<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Keyword arguments are used to mark the id<\/code> column as the primary key and False<\/code> as the default value for the complete<\/code> column.  Also, notice the __tablename__<\/code> field.  This is used to override the name of the table that maps to this class.  If you omit it, SQLAlchemy will generate a name based on the name of the model class.<\/p>\n\n\n\n
And that’s it for the SQLAlchemy model classes!<\/p>\n\n\n\n
Model Repository<\/h2>\n\n\n\n
The repository module contains functions that communicate with the database and handle converting between Pydantic schema classes and SQLAlchemy model classes.  Add a new Python file, repository.py<\/code> in the app<\/code> module folder.<\/p>\n\n\n\n
All database interaction is done via a SQLAlchemy Session<\/code>. And you’ll need the models<\/code> and schemas<\/code> too.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> sqlalchemy <\/span>import<\/span> select  <\/span><\/span>\nfrom<\/span> sqlalchemy<\/span>.<\/span>orm <\/span>import<\/span> Session<\/span><\/span>\nfrom<\/span> <\/span>.<\/span> <\/span>import<\/span> models<\/span>,<\/span> schemas<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The first function will use a Session<\/code> to retrieve all TodoItem<\/code>s from the database.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
def<\/span> <\/span>get_todo_items<\/span>(<\/span>sess<\/span>:<\/span> Session<\/span>):<\/span><\/span>\n  stmt <\/span>=<\/span> <\/span>select<\/span>(<\/span>models<\/span>.<\/span>TodoItem<\/span>)<\/span><\/span>\n  <\/span>return<\/span> sess<\/span>.<\/span>execute<\/span>(<\/span>stmt<\/span>).<\/span>scalars<\/span>().<\/span>all<\/span>()<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The next function gets a TodoItem<\/code> by its id<\/code>. It actually gets the first TodoItem<\/code> with the id<\/code>. But since the id<\/code> is the primary key it is unique and the first one is the only one.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
def<\/span> <\/span>get_todo_item_by_id<\/span>(<\/span>sess<\/span>:<\/span> Session<\/span>,<\/span> <\/span>todo_id<\/span>:<\/span> <\/span>int<\/span>):<\/span><\/span>\n  stmt <\/span>=<\/span> <\/span>select<\/span>(<\/span><\/span>\n    models<\/span>.<\/span>TodoItem<\/span><\/span>\n  <\/span>).<\/span>filter<\/span>(<\/span><\/span>\n    models<\/span>.<\/span>TodoItem<\/span>.<\/span>id <\/span>==<\/span> todo_id<\/span><\/span>\n  <\/span>)<\/span><\/span>\n  <\/span>return<\/span> sess<\/span>.<\/span>execute<\/span>(<\/span>stmt<\/span>).<\/span>scalar_one_or_none<\/span>()<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The create_todo_item<\/code> function also accepts a CreateTodoItem<\/code> schema. It uses the task<\/code> to create a new TodoItem<\/code> model object, then adds it to the Session and commits the change to the database. Before returning the new TodoItem<\/code> the Session refreshes it to make sure the id<\/code> value is included in the model.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
def<\/span> <\/span>create_todo_item<\/span>(<\/span><\/span>\n  <\/span>sess<\/span>:<\/span> Session<\/span>,<\/span> <\/span><\/span>\n  <\/span>create_todo<\/span>:<\/span> schemas<\/span>.<\/span>CreateTodoItem<\/span><\/span>\n):<\/span><\/span>\n  todo_item <\/span>=<\/span> models<\/span>.<\/span>TodoItem<\/span>(<\/span>task<\/span>=<\/span>create_todo<\/span>.<\/span>task<\/span>)<\/span><\/span>\n  sess<\/span>.<\/span>add<\/span>(<\/span>todo_item<\/span>)<\/span><\/span>\n  sess<\/span>.<\/span>commit<\/span>()<\/span><\/span>\n  sess<\/span>.<\/span>refresh<\/span>(<\/span>todo_item<\/span>)<\/span><\/span>\n  <\/span>return<\/span> todo_item<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The update_todo_item<\/code> function doesn’t need much explanation. First, use the get_todo_item_by_id<\/code> function to get the TodoItem<\/code>. Then use the UpdateTodoItem<\/code> schema to update the values in the TodoItem<\/code>. Then commit<\/code> and refresh<\/code> as in the create_todo_item<\/code> function.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> fastapi <\/span>import<\/span> HTTPException<\/span><\/span>\n<\/span>\ndef<\/span> <\/span>update_todo_item<\/span>(<\/span><\/span>\n  <\/span>sess<\/span>:<\/span> Session<\/span>,<\/span><\/span>\n  <\/span>todo_id<\/span>:<\/span> <\/span>int<\/span>,<\/span><\/span>\n  <\/span>update_todo<\/span>:<\/span> schemas<\/span>.<\/span>UpdateTodoItem<\/span><\/span>\n):<\/span><\/span>\n  todo_item <\/span>=<\/span> <\/span>get_todo_item_by_id<\/span>(<\/span>sess<\/span>,<\/span> todo_id<\/span>)<\/span><\/span>\n  <\/span>if<\/span> todo_item <\/span>is<\/span> <\/span>not<\/span> <\/span>None<\/span>:<\/span><\/span>\n    <\/span>if<\/span> update_todo<\/span>.<\/span>task <\/span>is<\/span> <\/span>not<\/span> <\/span>None<\/span>:<\/span><\/span>\n      todo_item<\/span>.<\/span>task <\/span>=<\/span> update_todo<\/span>.<\/span>task<\/span><\/span>\n    <\/span>if<\/span> update_todo<\/span>.<\/span>complete <\/span>is<\/span> <\/span>not<\/span> <\/span>None<\/span>:<\/span><\/span>\n      todo_item<\/span>.<\/span>complete <\/span>=<\/span> update_todo<\/span>.<\/span>complete<\/span><\/span>\n    sess<\/span>.<\/span>commit<\/span>()<\/span><\/span>\n    sess<\/span>.<\/span>refresh<\/span>(<\/span>todo_item<\/span>)<\/span><\/span>\n    <\/span>return<\/span> todo_item<\/span><\/span>\n  <\/span>raise<\/span> <\/span>HTTPException<\/span>(<\/span>status_code<\/span>=<\/span>404<\/span>,<\/span> <\/span>detail<\/span>=<\/span>"<\/span>Todo item not found<\/span>"<\/span>)<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Deleting an item is simple. Again, use the get_todo_item_by_id<\/code> function to get the TodoItem<\/code>. Call the delete<\/code> method on the Session and pass it the TodoItem<\/code>. Then commit the change. There is nothing to be refreshed as the TodoItem<\/code> was not modified.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
def<\/span> <\/span>delete_todo_item<\/span>(<\/span>sess<\/span>:<\/span> Session<\/span>,<\/span> <\/span>todo_id<\/span>:<\/span> <\/span>int<\/span>):<\/span><\/span>\n  todo_item <\/span>=<\/span> <\/span>get_todo_item_by_id<\/span>(<\/span>sess<\/span>,<\/span> todo_id<\/span>)<\/span><\/span>\n  <\/span>if<\/span> todo_item <\/span>is<\/span> <\/span>not<\/span> <\/span>None<\/span>:<\/span><\/span>\n    sess<\/span>.<\/span>delete<\/span>(<\/span>todo_item<\/span>)<\/span><\/span>\n    sess<\/span>.<\/span>commit<\/span>()<\/span><\/span>\n    <\/span>return<\/span> todo_item<\/span><\/span>\n  <\/span>raise<\/span> <\/span>HTTPException<\/span>(<\/span>status_code<\/span>=<\/span>404<\/span>,<\/span> <\/span>detail<\/span>=<\/span>"<\/span>Todo item not found<\/span>"<\/span>)<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
With the repository<\/code> module done we can move on to the FastAPI application in the main<\/code> module. But first, we need to finish up the db<\/code> module.<\/p>\n\n\n\n
Dependency Injection <\/h2>\n\n\n\n
Warning: this might get a little abstract because there is a lot of hand waving going with between Python and FastAPI.  But you don’t need to worry about it.  I’ll just go over the concepts.  <\/p>\n\n\n\n
Again, all interaction with a database happens via a SQLAlchemy Session. The sessionmaker<\/code> factory function was used in the db<\/code> module to create a session factory. But those sessions have to be managed. This is where dependency injection in FastAPI and generators in Python come into play.<\/p>\n\n\n\n
In the db<\/code> module, add a function to create and manage a SQLAlchemy Session.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
def<\/span> <\/span>get_db<\/span>():<\/span><\/span>\n  db <\/span>=<\/span> <\/span>Session<\/span>()<\/span><\/span>\n  <\/span>try<\/span>:<\/span><\/span>\n    <\/span>yield<\/span> db<\/span><\/span>\n  <\/span>finally<\/span>:<\/span><\/span>\n    db<\/span>.<\/span>close<\/span>()<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
On line 2, the Session factory creates a new SQLAlchemy Session. By convention this is called db<\/code>. Technically it is a connection to the database. Inside of the try<\/code> block on line 4 the Session is yielded meaning it is returned to the caller but the function does not return. As the yield<\/code> is inside of a try<\/code> block, when the yielded Session goes out of scope (ie. the request using it is done) control will return to the get_db<\/code> function. As there is no more code in the try<\/code> block, the finally<\/code> block will close<\/code> the Session.<\/p>\n\n\n\n
TLDR: you don’t have to worry about remembering to manually close the Session!<\/p>\n\n\n\n
It gets better. You don’t even have to call the get_db<\/code> function in the FastAPI handler function. Instead you use the dependency injection mechanism in FastAPI. When provided with a generator function (such as get_db<\/code>), FastAPI will make the yielded value available inside of the handler function. Let’s see how this works.<\/p>\n\n\n\n
FastAPI Application<\/h2>\n\n\n\n
First we need a number of imports. The FastAPI<\/code> application and the HTTPException<\/code> need to be imported from the fastapi<\/code> module. Also, from the fastapi<\/code> module, comes the Depends<\/code> function. In a minute you’ll see how this implements dependency injection with the get_db<\/code> function that is imported from the db<\/code> module. And for type hints, bring in the Session<\/code> class from SQLAlchemy. To do the work of the database, we need the models<\/code>, schemas<\/code> and repository<\/code> modules.<\/p>\n\n\n\n
And don’t forget to create a FastAPI application!<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> fastapi <\/span>import<\/span> FastAPI<\/span>,<\/span> Depends<\/span>,<\/span> HTTPException<\/span><\/span>\nfrom<\/span> sqlalchemy<\/span>.<\/span>orm <\/span>import<\/span> Session<\/span><\/span>\n<\/span>\nfrom<\/span> <\/span>.<\/span> <\/span>import<\/span> models<\/span>,<\/span> schemas<\/span>,<\/span> repository<\/span><\/span>\nfrom<\/span> <\/span>.<\/span>db <\/span>import<\/span> get_db<\/span><\/span>\n<\/span>\napp <\/span>=<\/span> <\/span>FastAPI<\/span>()<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The first endpoint will get all TodoItem<\/code>s from the database.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> typing <\/span>import<\/span> List<\/span><\/span>\n<\/span>\n@<\/span>app<\/span>.<\/span>get<\/span>(<\/span>"<\/span>\/todos<\/span>"<\/span>,<\/span> <\/span>response_model<\/span>=<\/span>List<\/span>[<\/span>schemas<\/span>.<\/span>TodoItem<\/span>])<\/span><\/span>\ndef<\/span> <\/span>get_todo_items<\/span>(<\/span>db<\/span>:<\/span> Session <\/span>=<\/span> <\/span>Depends<\/span>(<\/span>get_db<\/span>)):<\/span><\/span>\n  <\/span>return<\/span> repository<\/span>.<\/span>get_todo_items<\/span>(<\/span>db<\/span>)<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Here we see the first use of the Depends<\/code> function. The get_todo_items<\/code> repository function needs a SQLAlchemy Session<\/code>. The handler function accepts one. But we don’t call that function explicitly, FastAPI does in response to an HTTP request. Therefore, the Depends<\/code> function calls a factory function and passes the yielded value to the handler function. In other words, Depends<\/code> calls get_db<\/code> and passes the yielded Session<\/code> to the get_todo_item<\/code> handler function as the db<\/code> parameter. Inside of the handler function, the get_todo_items<\/code> repository function uses the Session<\/code> to query the database. When the get_todo_items<\/code> repository function returns, the finally<\/code> block in the get_db<\/code> function will close<\/code> the Session<\/code> acquired via the Depends<\/code> DI function.<\/p>\n\n\n\n
TLDR: you don’t have to worry about remembering to manually close the database Session<\/code>!<\/p>\n\n\n\n
Looking at the rest of the handler functions, you’ll notice they follow the same pattern. Get a database Session<\/code> via dependency injection, use it to call a repository function, return the results to the caller or raise an exception. Here is the handler function to get a TodoItem<\/code> by id<\/code>.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
@<\/span>app<\/span>.<\/span>get<\/span>(<\/span>"<\/span>\/todos\/<\/span>{todo_id}<\/span>"<\/span>,<\/span> <\/span>response_model<\/span>=<\/span>schemas<\/span>.<\/span>TodoItem<\/span>)<\/span><\/span>\ndef<\/span> <\/span>get_todo_item_by_id<\/span>(<\/span>todo_id<\/span>:<\/span> <\/span>int<\/span>,<\/span> <\/span>db<\/span>:<\/span> Session <\/span>=<\/span> <\/span>Depends<\/span>(<\/span>get_db<\/span>)):<\/span><\/span>\n  todo_item <\/span>=<\/span> repository<\/span>.<\/span>get_todo_item_by_id<\/span>(<\/span>db<\/span>,<\/span> todo_id<\/span>)<\/span><\/span>\n  <\/span>if<\/span> todo_item <\/span>is<\/span> <\/span>None<\/span>:<\/span><\/span>\n    <\/span>raise<\/span> <\/span>HTTPException<\/span>(<\/span>status_code<\/span>=<\/span>404<\/span>,<\/span> <\/span>detail<\/span>=<\/span>"<\/span>TodoItem not found<\/span>"<\/span>)<\/span><\/span>\n  <\/span>return<\/span> todo_item<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
And the remaining function handlers should be self explanatory given the context of the previous post.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
@<\/span>app<\/span>.<\/span>post<\/span>(<\/span>"<\/span>\/todos<\/span>"<\/span>,<\/span> <\/span>response_model<\/span>=<\/span>schemas<\/span>.<\/span>TodoItem<\/span>)<\/span><\/span>\ndef<\/span> <\/span>create_todo_item<\/span>(<\/span><\/span>\n  <\/span>create_todo<\/span>:<\/span> schemas<\/span>.<\/span>CreateTodoItem<\/span>,<\/span> <\/span><\/span>\n  <\/span>db<\/span>:<\/span> Session <\/span>=<\/span> <\/span>Depends<\/span>(<\/span>get_db<\/span>)<\/span><\/span>\n):<\/span><\/span>\n  <\/span>return<\/span> repository<\/span>.<\/span>create_todo_item<\/span>(<\/span>db<\/span>,<\/span> create_todo<\/span>)<\/span><\/span>\n  <\/span><\/span>\n@<\/span>app<\/span>.<\/span>put<\/span>(<\/span>"<\/span>\/todos\/<\/span>{todo_id}<\/span>"<\/span>,<\/span> <\/span>response_model<\/span>=<\/span>schemas<\/span>.<\/span>TodoItem<\/span>)<\/span><\/span>\ndef<\/span> <\/span>update_todo_item<\/span>(<\/span><\/span>\n  <\/span>todo_id<\/span>:<\/span> <\/span>int<\/span>,<\/span> <\/span><\/span>\n  <\/span>update_todo<\/span>:<\/span> schemas<\/span>.<\/span>UpdateTodoItem<\/span>,<\/span> <\/span><\/span>\n  <\/span>db<\/span>:<\/span> Session <\/span>=<\/span> <\/span>Depends<\/span>(<\/span>get_db<\/span>)<\/span><\/span>\n):<\/span><\/span>\n  updated_todo_item <\/span>=<\/span> repository<\/span>.<\/span>update_todo_item<\/span>(<\/span><\/span>\n    db<\/span>,<\/span> <\/span><\/span>\n    todo_id<\/span>,<\/span> update_todo<\/span><\/span>\n  <\/span>)<\/span><\/span>\n  <\/span>if<\/span> updated_todo_item <\/span>is<\/span> <\/span>None<\/span>:<\/span><\/span>\n    <\/span>raise<\/span> <\/span>HTTPException<\/span>(<\/span>status_code<\/span>=<\/span>404<\/span>,<\/span> <\/span>detail<\/span>=<\/span>"<\/span>TodoItem not found<\/span>"<\/span>)<\/span><\/span>\n  <\/span>return<\/span> updated_todo_item<\/span><\/span>\n  <\/span><\/span>\n@<\/span>app<\/span>.<\/span>delete<\/span>(<\/span>"<\/span>\/todos\/<\/span>{todo_id}<\/span>"<\/span>,<\/span> <\/span>status_code<\/span>=<\/span>204<\/span>)<\/span><\/span>\ndef<\/span> <\/span>delete_todo_item<\/span>(<\/span>todo_id<\/span>:<\/span> <\/span>int<\/span>,<\/span> <\/span>db<\/span>:<\/span> Session <\/span>=<\/span> <\/span>Depends<\/span>(<\/span>get_db<\/span>)):<\/span><\/span>\n  deleted_todo_item <\/span>=<\/span> repository<\/span>.<\/span>delete_todo_item<\/span>(<\/span>todo_id<\/span>,<\/span> db<\/span>)<\/span><\/span>\n  <\/span>if<\/span> deleted_todo_item <\/span>is<\/span> <\/span>None<\/span>:<\/span><\/span>\n    <\/span>raise<\/span> <\/span>HTTPException<\/span>(<\/span>status_code<\/span>=<\/span>404<\/span>,<\/span> <\/span>detail<\/span>=<\/span>"<\/span>TodoItem not found<\/span>"<\/span>)<\/span><\/span>\n  <\/span>return<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The application is ready to run … except for one thing.<\/p>\n\n\n\n
Database Migrations with Alembic<\/h2>\n\n\n\n
There is no database!  Let’s fix that.<\/p>\n\n\n\n
It is possible to create the database in code.  Given an instance of a SQLAlchemy Engine<\/code> you could do this in the main<\/code> module.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
Base<\/span>.<\/span>metadata<\/span>.<\/span>create_all<\/span>(<\/span>bind<\/span>=<\/span>engine<\/span>)<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
And it will generate a database with a table for each model class deriving from Base<\/code>.  But when you start making changes to the data model, this method can cause issues.  A better choice is to use migrations.  Alembic is a tool that will detect changes in the data model of your application and generate Python scripts to manage those changes.  Each change is called a migration.  If you find that a migration doesn’t work, you can roll back to a previous one.  It might sound like a lot of work and on the part of Alembic it is<\/strong> a lot of work but it’s work you don’t have to do.<\/p>\n\n\n\n
Make sure that you installed the alembic<\/code> package.  This will install a command line app that you run to scaffold out some code Alembic needs for configuration.  Run the command in the project root (the parent folder of the app<\/code> module).<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
$<\/span> <\/span>alembic<\/span> <\/span>init<\/span> <\/span>alembic<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
The alembic<\/code> command will create an alembic.ini <\/code>file and a folder named alembic<\/code>.  You’ll need to modify two files that were generated by alembic<\/code>.  First in alembic.ini<\/code>, find the sqlalchemy.url<\/code> key and set it to the connection string to the SQLite database.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
sqlalchemy.url<\/span> <\/span>=<\/span> sqlite:\/\/\/.\/blog.db<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Next, in env.py,<\/code> in the alembic<\/code> folder, import the Base<\/code> SQLAlchemy model class from the app.db<\/code> module, and the models<\/code> module itself.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
from<\/span> app<\/span>.<\/span>db <\/span>import<\/span> Base<\/span><\/span>\nfrom<\/span> app <\/span>import<\/span> models<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Finally, in env.py<\/code>, find the target_metadata<\/code> variable and set it to Base.metadata<\/code>.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
target_metadata <\/span>=<\/span> Base<\/span>.<\/span>metadata<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Save everything and go back to the command line to create your first migration.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
$<\/span> <\/span>alembic<\/span> <\/span>revision<\/span> <\/span>--autogenerate<\/span> <\/span>-m<\/span> <\/span>"<\/span>Initial migration<\/span>"<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
This will create a new migration with the description “Initial migration”.   In the alembic<\/code> folder, in the versions<\/code> folder, a new Python script has been generated to add the todo_items<\/code> table to the database which will map to the TodoItem<\/code> SQLAlchemy model.<\/p>\n\n\n\n
To apply the migration run the upgrade<\/code> command.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
$<\/span> <\/span>alembic<\/span> <\/span>upgrade<\/span> <\/span>head<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Notice in the project root (alongside the app<\/code> and alembic<\/code> folders) is a new blog.db<\/code> SQLite database.<\/p>\n\n\n\n
Take It For a Spin<\/h2>\n\n\n\n
Now we’re ready to run the application.  From the project root, start the application with uvicorn.<\/p>\n\n\n\n
<\/circle><\/circle><\/circle><\/g><\/svg><\/span><\/path><\/path><\/svg><\/span>
$<\/span> <\/span>uvicorn<\/span> <\/span>app.main:app<\/span> <\/span>--reload<\/span><\/span><\/code><\/pre><\/div>\n\n\n\n
Navigate to http:\/\/127.0.0.1\/docs in the browser.  You should see five endpoints.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Expand the POST endpoint.  Click the Try it out<\/strong> button and provide a task in the JSON payload.  Click the Execute<\/strong> button and scroll down to see the JSON for the new TodoItem<\/code> in the database.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Expand the GET endpoint to get all TodoItem<\/code>s and execute it.  The TodoItem<\/code> you just created is returned.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Now expand the PUT endpoint.  Enter 1 for the todo_id<\/code>.  In the request JSON, remove the task<\/code> key and set the complete<\/code> key to true<\/code>.  Execute the request and notice in the response JSON that complete<\/code> is now true<\/code>.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Go back to the GET endpoints to get all TodoItem<\/code>s and notice it has been updated in the database.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Expand the DELETE endpoint.  Enter 1 in the todo_id<\/code> input box, and execute it.  The response is a 204 status code with no body content.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
For a sanity check, use the GET endpoint to get all TodoItem<\/code>s and the result should be an empty JSON array as the only item in the database was deleted.<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
Summary<\/h2>\n\n\n\n
in this post you saw how to use SQLAlchemy to integrate FastAPI with a SQLite database.  You saw how to create SQLAlchemy model classes and map them to Pydantic schema classes.  To manage SQLAlchemy Sessions, you used FastAPI’s dependency injection.  And you used Alembic to manage changes to the data model and sync them to the database.<\/p>\n","protected":false},"excerpt":{"rendered":"
In the previous post, you saw how to implement a complete REST API. To keep…<\/p>\n","protected":false},"author":1,"featured_media":91,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[5],"tags":[12,3,13,4,11],"class_list":["post-71","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-fastapi","tag-alembic","tag-fastapi","tag-pydantic","tag-python","tag-sqlalchemy"],"_links":{"self":[{"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/posts\/71","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/comments?post=71"}],"version-history":[{"count":18,"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/posts\/71\/revisions"}],"predecessor-version":[{"id":154,"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/posts\/71\/revisions\/154"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/media\/91"}],"wp:attachment":[{"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/media?parent=71"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/categories?post=71"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/douglasstarnes.dev\/index.php\/wp-json\/wp\/v2\/tags?post=71"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}