Dolt is the world’s first version-controlled SQL database. It’s also a drop-in replacement for MySQL. This means that Dolt works anywhere MySQL works. We’ve written several blogs showing off Dolt’s compatibility with other libraries and frameworks, but we haven’t yet covered FastAPI.
FastAPI is a popular Python framework for building lightweight, performant APIs. It’s also relatively easy to use, relying on Python type hints for request/response validation and automatic doc generation based on the OpenAPI standard. In this article, we’ll use the framework to build a simple API and integrate it with a Dolt database.
I wrote extensively about the sample application we’ll build in this blog, so make sure to read that first to get a better idea of what the app can do. In short, it’s a list making tool with an AI component that enables an agent to make changes to your list. This article will focus on the FastAPI backend that handles CRUD operations on lists as well as Dolt’s branch, diff, and merge functionality. Let’s get started.
Prerequisites#
First, you’ll need to install Dolt. You can find installation instructions here. I’d also recommend setting up a Python virtual environment. This isn’t strictly necessary, but it’s good practice. I’ll start by creating the project directory.
mkdir dolt-lists
cd dolt-listsThen, create the virtual environment with:
python -m venv .venvAnd activate it by running:
source .venv/bin/activateWith the virtual environment active, we can now install FastAPI.
pip install "fastapi[standard]"Now, we’re ready to go. If you’ve never used FastAPI before, it also might be worth skimming through the tutorial on the official website, which covers all the basic syntax and quirks of the framework.
Database Setup#
Let’s start by creating the Dolt database and defining the data models. First, you’ll want to navigate to the directory where you want your database to live, then run:
dolt sql-serverThis command spins up a Dolt SQL server running on localhost port 3306. Now, in another terminal, connect to the server by running:
dolt sqlThis will bring up a SQL client allowing you to interact with the server. Let’s create the database.
CREATE DATABASE lists;
USE lists;And then create the tables:
CREATE TABLE `lists` (
`name` varchar(255) NOT NULL,
PRIMARY KEY (`name`)
)
CREATE TABLE `list_items` (
`id` int NOT NULL AUTO_INCREMENT,
`item` text NOT NULL,
PRIMARY KEY (`id`)
)We’ll also create a SQL user for the application:
GRANT ALL PRIVILEGES ON lists.* TO 'list_app'@'%' IDENTIFIED BY 'list_app_pass';Finally, we’ll make a Dolt commit to officially track our new tables.
CALL DOLT_COMMIT('-Am', 'Create `lists` and `list_items` tables')
Now our working set is clean and we have a fresh database for our application to operate on.
Mapping to FastAPI#
Now, with our tables defined, we’re ready to translate these schemas to data models in our application. FastAPI uses the SQLModel ORM, which is built on top of SQLAlchemy specifically for FastAPI. To start, we’ll define Python classes that correspond to our database tables. The first one is for the lists table.
class DoltList(SQLModel, table=True):
__tablename__ = "lists"
name: str = Field(primary_key=True, sa_type=Text)Let’s walk through what’s happening here. We start by creating a class that inherits from the SQLModel base class, and table=True means that the class we’re defining represents a table in our database. The __tablename__ variable is necessary if the name of your database table differs from the name of your class. Finally, we can define columns using the Field class. Here, we’re letting SQLModel know that the name column is the primary key and that the SQL type should be TEXT. This is the SQLModel class for the list_items table:
class DoltListItem(SQLModel, table=True):
__tablename__ = "list_items"
id: Optional[int] = Field(default=None, primary_key=True)
item: str = Field(sa_type=Text)Here, we have an integer ID primary key. Since id is the only primary key column and we’re explicitly saying default=None, SQLModel interprets this to mean that the id field should autoincrement. With our data models defined, let’s now connect to the database from our application. I created a file called db.py with the following:
DB_URL = os.getenv(
"DATABASE_URL",
"mysql+mysqlconnector://list_app:list_app_pass@127.0.0.1:3306/lists/main"
)
engine = create_engine(DB_URL, echo=False, pool_pre_ping=True)
def get_session():
with Session(engine) as session:
yield sessionAgain, since Dolt is fully MySQL-compatible, we can use the Python MySQL connector to connect to the database. We then use SQLModel’s create_engine() method to get a global reference to the database engine object, and the get_session function creates a new SQL session from the engine. We’ll use this function throughout the application to interact with our database.
Creating Endpoints#
We’re finally ready for our application to actually do something. First, we’ll create a separate file called main.py where we’ll instantiate the FastAPI app object and set up an API router.
app = FastAPI()
router = APIRouter(prefix="/api/v1")
# All your FastAPI endpoints ...
app.include_router(router)The router prefixes all API routes with “/api/v1”. It’s important to include the app.include_router() line after you’ve defined your FastAPI endpoints so that Python has references to them. Let’s create our first API endpoint for creating a new list:
@router.post("/lists", response_model=CreateListResponse)
def create_list(list_req: CreateListRequest, session: Session = Depends(checkout_and_get_session)):
new_branch = checkout_branch(session, new=True)
new_list = DoltList(name=list_req.name)
session.add(new_list)
session.commit()
dolt_commit(session, new_branch)
session.refresh(new_list)
return CreateListResponse(branch=new_branch, name=new_list.name)There’s a lot going on here, so let’s take it step by step. FastAPI endpoints are defined with function decorators using the router we defined previously. The formula for these looks like this:
@router.HTTP_METHOD("/my/api/route", response_model=MyResponseModelClass)FastAPI automatically handles serialization for your request and response classes. For this endpoint, these are defined as follows:
class CreateListRequest(BaseModel):
name: str
class CreateListResponse(BaseModel):
branch: str
name: strAny request/response classes should inherit from FastAPI’s BaseModel class. We also get access to the database session from a function parameter using a FastAPI dependency. The following expression says that before this function can run, you must first get access to a database Session object by calling the checkout_and_get_session() function.
session: Session = Depends(checkout_and_get_session)In our app, every list lives on its own branch. As such, it’s necessary that the SQL session corresponds to the correct branch. This is the checkout_and_get_session() function:
def checkout_and_get_session(
list_branch: Optional[str] = "main",
session: Session = Depends(get_session)
) -> Session:
checkout_branch(session, branch=list_branch)
return sessionIt takes an optional list_branch argument, defaulting to “main”, then switches to that branch on the session and returns the session object. This is the checkout_branch() logic:
def checkout_branch(session: Session, branch: str = "main", new: bool = False):
branch_name = str(uuid4()) if new else branch
try:
if new:
session.exec(
text("CALL DOLT_BRANCH(:branch)"),
params={"branch": branch_name}
)
session.exec(
text(f"USE `lists/{branch_name}`")
)
session.commit()
except Exception as e:
if "did not match any table(s) known to dolt" in str(e):
raise HTTPException(status_code=404, detail="List not found")
raise
return branch_nameThis function optionally creates a new branch using the CALL DOLT_BRANCH("branch_name") stored procedure. In FastAPI, you can execute SQL statements using the session.exec() method. As we’ll see later, you can use built-in SQL builder functionality to query against your ORM models. In this case, however, we’re calling a stored procedure, so we need to directly write the SQL statement with its parameters. For our create_list endpoint, we initially want to make sure we’re on the main branch, so checkout_branch() will simply execute USE lists/main to switch to main. Branches are super versatile in Dolt, and there are several ways to switch between them. When working with ORMs and web frameworks that manage connection pools, we recommend using the USE database/branch syntax for handling branch switches. For more on that topic, check out this blog.
Great, now that we’re on the main branch, we’ll create our list by first creating a branch for it. To do this, we’ll directly call checkout_branch(), but this time with new=True. This will create a new branch off main with a UUID, and then switch to it. Now that we’re on the list branch, we’ll create the actual list by creating a DoltList object with the desired list name, then calling session.add() with the list object. This will run a SQL insert query on the new branch, inserting the new row into the lists table. Since Python disables autocommit by default, we then call session.commit() to complete the transaction.
After this is done, we also want to make Dolt commit. To do this, we call the dolt_commit function, which looks like this:
def dolt_commit(session: Session, branch_name):
changed_tables = session.exec(
text("SELECT table_name FROM DOLT_STATUS")
).scalars().all()
if changed_tables:
commit_message = f"Changes to {', '.join(changed_tables)} from {branch_name}"
session.exec(
text("CALL DOLT_COMMIT(:flag, :commit_message)"),
params={"flag": "-Am", "commit_message": commit_message}
)
session.commit()This first executes a select statement on the DOLT_STATUS system table to determine what tables have changed in your working set. This isn’t strictly necessary, but it’s nice to have more informative commit messages. Then, we run the CALL DOLT_COMMIT() stored procedure to actually make the commit.
This is the basic flow for most of our app’s functionality. Switch to the correct branch, make changes to the database, then commit. Let’s look at a few more examples. Here’s the endpoint for fetching a list:
@router.get("/lists/{list_branch}", response_model=GetListResponse)
def get_list(session: Session = Depends(checkout_and_get_session)):
list_name = session.exec(select(DoltList.name)).one()
items = session.exec(select(DoltListItem)).all()
return GetListResponse(list_name=list_name, items=items)Here, we’re including list_branch as a path variable, so when checkout_and_get_session is called, it will use the path variable to switch to list_branch instead of defaulting to main. In this endpoint, we want to return the name of the list as well the items that belong to it. Since this information all comes from tables that correspond to our ORM models, we can use SQLModel’s select() function with our table classes to more easily build SQL queries. For example, session.exec(select(DoltListItem)).all() translates to SELECT * FROM list_items.
Here’s another endpoint that updates a specific list item:
@router.patch("/lists/{list_branch}/items/{item_id}", response_model=DoltListItem)
def update_list_item(
list_branch: str,
item_id: int,
item_req:
ListItemRequest,
session: Session = Depends(checkout_and_get_session)
):
item = session.get(DoltListItem, item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
if item_req.item is not None:
item.item = item_req.item
session.add(item)
session.commit()
dolt_commit(session, list_branch)
session.refresh(item)
return itemWe first switch to the correct branch, then fetch the existing item using session.get() with the DoltListItem class and the item_id. After doing some validations, we set the item field on the object to the item passed in the request, then session.add() and session.commit(). After making a Dolt commit, we then call session.refresh() on the object, which is a useful method to ensure that your in-memory representation of an ORM object matches that of your database.
More Dolt-Specific Operations#
Our sample application also exposes a few endpoints that make use of some Dolt-specific operations, specifically diff and merge. This is the endpoint that powers the list diff view in our application:
@router.get("/diff", response_model=GetDiffResponse)
def get_diff(from_branch: str, to_branch: str, session: Session = Depends(get_session)):
return dolt_diff(session, from_branch, to_branch)Notice that from_branch and to_branch are not path variables here. Instead, FastAPI interprets any parameters that don’t match a path variable as query parameters (i.e. you would call this endpoint with /api/v1/diff?from_branch=branch1&to_branch=branch2). Also, we aren’t using the checkout_and_get_session dependency here because Dolt allows you to diff two branches regardless of the branch you’re currently on. This is how the dolt_diff() function works:
def dolt_diff(session: Session, from_branch: str, to_branch: str) -> GetDiffResponse:
list_result = session.exec(
text("SELECT to_name, diff_type FROM DOLT_DIFF(:from_branch, :to_branch, :table_name)"),
params={"from_branch": from_branch, "to_branch": to_branch, "table_name": "lists"}
).first()
if list_result:
to_list_name, list_name_diff_type = list_result
else:
to_list_name, list_name_diff_type = None, None
diff_rows = session.exec(
text("SELECT from_id, to_id, to_item, diff_type FROM DOLT_DIFF(:from_branch, :to_branch, :table_name)"),
params={"from_branch": from_branch, "to_branch": to_branch, "table_name": "list_items"}
).all()
item_diffs = []
for from_id, to_id, to_item, diff_type in diff_rows:
item_diffs.append(ItemDiff(from_id=from_id, to_id=to_id, to_item=to_item, item_diff_type=diff_type))
return GetDiffResponse(
to_list_name=to_list_name,
list_name_diff_type=list_name_diff_type,
item_diffs=item_diffs
)We start by using the DOLT_DIFF() table function on the lists table to determine if the list name has changed between branches. Then, we unwrap the result into separate variables. The diff_type field can be either “added”, “modified”, or “removed”. Next, we use DOLT_DIFF() again, but this time on the list_items table to find all the differences between individual items. All this information is then wrapped into the GetDiffResponse class and returned to the caller.
Next, let’s look at the endpoint that handles merges:
@router.post("/agent/change/{list_branch}", status_code=200)
def handle_agent_changes(agent_change_req: AgentChangeRequest, session: Session = Depends(checkout_and_get_session)):
if agent_change_req.accept:
dolt_merge(session, agent_change_req.change_branch)
delete_branch(session, agent_change_req.change_branch)First, we switch to the user’s branch, which will be the base branch for the merge. Then, if the user decided to accept the changes, we call dolt_merge:
def dolt_merge(session: Session, change_branch: str):
session.exec(
text("CALL DOLT_MERGE(:change_branch)"),
params={"change_branch": change_branch}
)
session.commit()This executes the DOLT_MERGE() stored procedure, which applies all the changes from change_branch to the current branch. Finally, we delete the branch:
def delete_branch(session: Session, branch_to_delete: str):
session.exec(
text("CALL DOLT_BRANCH(:flag, :branch)"),
params={"flag": "-D", "branch": branch_to_delete}
)
session.commit()Conclusion#
Our sample application involves a lot of very Dolt-specific functionality, and FastAPI made it fairly easy to integrate those features into the app. Feel free to check out the code for this project here. If you have any questions about using FastAPI with Dolt, or if you’re curious about Dolt’s compatibility with a framework we haven’t covered yet, come by our Discord and let us know.