WARNING
After migrating a database with a migration file, make sure to commit to a Git
repository. Avoid re-editing migration files after migrating a database as Alembic will
skip existing migrations by cross-checking them with its versioning table.
If a migration file has already been run, it won’t detect changes in its content.
To update your database schema, create a new migration file instead.
Following the aforementioned workflow will now allow you to
not only version control your database schemas but also
manage changes to your production environments as your
application requirements change.
Storing Data When Working with
Real-Time Streams
You should now be in a position to implement your own CRUD
endpoints to retrieve and mutate both user conversation and
message records in your database.
One question that remains unanswered is how to handle
transactions within data streaming endpoints, such as an LLM
streaming outputs to a client.
You can’t stream data into a traditional relational database as
ensuring ACID compliance with streaming transactions will
prove challenging. Instead, you will want to perform your
standard database operation as soon as your FastAPI server
returns a response to the client. This challenge is exactly what a
FastAPI’s background task can solve, as you can see in
Example 7-15.
Example 7-15. Storing content of an LLM output stream
# [Link]
from itertools import tee
from database import DBSessionDep
from entities import Message
from fastapi import BackgroundTasks, Depends
from [Link] import StreamingResponse
from [Link] import Conversati
from [Link] import MessageReposito
from [Link] import AsyncSession
async def store_message(
prompt_content: str,
response_content: str,
conversation_id: int,
session: AsyncSession,
) -> None:
message = Message(
conversation_id=conversation_id,
prompt_content=prompt_content,
response_content=response_content,
)
await MessageRepository(session).create(messa
@[Link]("/text/generate/stream")
async def stream_llm_controller(
prompt: str,
background_task: BackgroundTasks,
session: DBSessionDep,
conversation: Conversation = Depends(get_conv
) -> StreamingResponse:
# Invoke LLM and obtain the response stream
...
stream_1, stream_2 = tee(response_stream)
background_task.add_task(
store_message, prompt, "".join(stream_1)
)
return StreamingResponse(stream_2)
Create a function to store a message against a
conversation.
Check that the conversation record exists and fetch it
within a dependency.
Create two separate copies of the LLM stream, one for the
StreamingResponse and another to process in a
background task.
Create a background task to store the message after the
StreamingResponse is finished.
In Example 7-15, you allow FastAPI to fully stream the LLM
response to the client.
It won’t matter whether you’re using an SSE or WebSocket
endpoint. Once a request a response is fully streamed, invoke a
background task passing in the full stream response content.
Within the background task, you can then run a function to
store the message after the request is sent, with the full LLM
response content.
Using the same approach, you can even generate a title for a
conversation based on the content of the first message. To do
this, you can invoke the LLM again with the content of the first
message in the conversation, requesting for an appropriate title
for the conversation. Once a conversation title is generated, you
can create the conversation record in the database, as shown in
Example 7-16.
Example 7-16. Using the LLM to generate conversation titles
based on the initial user prompt
from entities import Conversation
from openai import AsyncClient
from [Link] import Conversati
from [Link] import AsyncSession
async_client = AsyncClient(...)
async def create_conversation(
initial_prompt: str, session: AsyncSession
) -> Conversation:
completion = await async_client.[Link]
messages=[
{
"role": "system",
"content": "Suggest a title for t
"based on the user pro
},
{
"role": "user",
"content": initial_prompt,
},
],
model="gpt-3.5-turbo",
)
title = [Link][0].[Link]
conversation = Conversation(
title=title,
# add other conversation properties
...
)
return await ConversationRepository(session)
Using SQLAlchemy with Alembic is a tried and tested approach
to working with relational databases in FastAPI, so you’re more
likely to find a lot of resources on integrating these
technologies.
Both the SQLAlchemy ORM and Alembic allow you to interact
with your database and control the changes to its schemas.
Summary
In this chapter, you dove into the critical aspects of integrating a
database into your FastAPI application to store and retrieve
user conversations.
You learned to identify when a database is necessary and how
to identify the appropriate type for your project, whether it be
relational or nonrelational. By understanding the underlying
mechanisms of relational databases and the use cases for
nonrelational databases, you’re now equipped to make
informed decisions about database selection.
You also explored the development workflow, tooling, and best
practices for working with relational databases. This includes
learning techniques to improve query performance and
efficiency, as well as strategies for managing evolving database
schema changes. Additionally, you gained insights into
managing codebase, database schema, and data drifts when
working in teams.
As you move forward, the next chapter will guide you through
implementing user management, authentication, and
authorization mechanisms. This will further enhance your
application’s security and user experience, building on the solid
database foundation you’ve established in this chapter.
1
Refer to this reddit discussion thread.
Part III. Securing, Optimizing, Testing,
and Deploying AI Services
In this part, you will learn about security, optimizing, testing,
and deployment best practices when building GenAI services.
Chapter 8. Authentication and
Authorization
CHAPTER GOALS
In this chapter, you will learn about:
Relevant authentication strategies for securing your GenAI
services
How to implement basic credentials and JSON Web Token
(JWT) authentication from scratch
Various authentication risks and attack vectors
How to build single sign-on (SSO) authentication with an
identity provider such as GitHub following the OAuth2
standard
Authorization patterns such as role-based, attribute-based,
and relationship-based access control
Authorization mechanisms to safeguard resources from
nonprivileged users
How to restrict access to resources and limit AI generation
outputs based on user privileges
So far, you’ve built GenAI services that can interact with
databases, stream model responses, and handle concurrent
users.
Your services are now up and running, but since they’re not
protected from attackers or malicious users, deploying them to
production may prove problematic.