Learning FastA
I've been procrastinating a lot about trying to learn a proper modern backend framework, but I guess now's the time. Why FastAPI? IDK it's just nice I guess. Anyways, here we are documenting the entire process, of course for my own reference later.
Setting up the environments and other stuff
The first step is creating a virtual environment and installing all the dependencies. For the dependencies, use the below command:
uv add python-dotenv fastapi-users[sqlalchemy] imagekitio uvicorn[standard] aiosqlite
Next step is setting up the ImageKit environment variables. Go to the ImageKit website, register, you'll find both the private and public keys, add them to your environment along with the custom ImageKit URL endpoint.
Test things out with a dummy endpoint
Now that we're done with the basic setup, we'll be testing things out with a simple hello world get route. Here's the dummy endpoint code for the app/app.py file:
from fastapi import FastAPI
app = FastAPI()
@app.get("/hello-world")
def hello_world():
return {"message": "Hello, World!"}
Next, I set up the main.py file using uvicorn.
import uvicorn
if __name__ == "__main__":
uvicorn.run("app.app:app", host="0.0.0.0", port=8000, reload=True)
Now we can just run the server using "uv run ./main.py".
Using HTTPExceptions and Path Parameters
This is another cool package that let's you implement exceptions in your code, I'm not explaining since the current me and the users know about them. Additionally, path parameters are used to pass specific parameters for filtering inside the path.
Here's the code snippet to show how that works:
from fastapi import FastAPI, HTTPException
app = FastAPI()
text_post = {
1: {"title": "First Post", "content": "Cool first post",},
2: {"title": "Second Post", "content": "Cool second post",},
}
@app.get("/{id}")
def fetch_one_post(id: int):
if id not in text_post:
return HTTPException(status_code=404, detail="Post not found")
return text_post.get(id)
Using Query Parameters
Query parameters are basically used for filtering out content. For example if we want to limit a number of outputs from the path like "your-project-path/all_posts?limit=5". Below is an example code snippet the get these things clear.
from fastapi import FastAPI, HTTPException
app = FastAPI()
text_post = {...}
@app.get("/")
def fetch_all_posts(limit: int = None):
if limit is not None:
return dict(list(text_post.items())[:limit])
return text_post
Setup the database
For now, we'll be creating a simple async database and I'll explain each part of the code.
from collections.abc import AsyncGenerator
import uuid
from sqlalchemy import Column, String, Text, DateTime, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, relationship
import datetime
DATABASE_URL = "sqlite+aiosqkite:///./test.db"
First we define a database URL. This URL will be used as a connection string. The general structure of the URL can be generalized to:
dialect+driver://username:password@host:port/database
here,
sqlite: database type (dialect)
aiosqlite: Async driver
./test_db: local file
In case of a PostgreSQL database, the URL would look like:
DATABASE_URL = ("postgresql+asyncpg://postgres:password@localhost:5432/social_db")
This means:
postgresql: Dialect/Database
asyncpg: Driver
postgres: Username
password: Password
localhost: Server
5432: Port
social_db: Database name
class Base(DeclarativeBase):
pass
class Post(Base):
__tablename__ = "posts"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
caption = Column(Text)
url = Column(String, nullable=False)
file_type = Column(String, nullable=False)
file_name = Column(String, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
Define the Post schema. Next up is creating the engine and the session maker.
engine = create_async_engine(DATABASE_URL, echo=True)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)
The engine is responsible for opening database connections, managing connection pools, sending SQL queries, Receiving results and handling transactions.
The session maker, also called the session factory is responsible for tracking the new, updated and deleted objects. Basically any change made to the database will be first tracked by the session factory. Once it commits those changes, SQLAlchemy generates respective SQL commands are generated which are then passed to the database.
expire_on_commit=_
Basically, after every commit you need to decide whether the data needs to be removed or not. If this is set to True, the SQLAlchemy might need to reload the data from the database.
async def create_db_and_tables():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session_maker() as session:
yield session
Define the functions for initializing the database and the session. Then we'll be making some changes to the app.py file:
from app.db import create_db_and_tables, get_async_session
from sqlalchemy.ext.asyncio import AsyncSession
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
await create_db_and_tables()
yield
app = FastAPI(lifespan=lifespan)
This allows FastAPI to automatically manage sessions and initialize the database everytime the app starts.
Setting up the routes
The first route we set-up would be the file upload route. Before defining the actual route, we'll first configure ImageKit. In app/images.py, initialize ImageKit and it will automatically fetch the API keys.
from imagekitio import ImageKit
imagekit = ImageKit()
Now, let's define the post route.
@app.post("/upload")
async def upload_file(
file: UploadFile = File(...),
caption: str = Form(...),
session: AsyncSession = Depends(get_async_session)):
try:
response = imagekit.files.upload(
file=file.file,
file_name=file.filename,
use_unique_file_name=True,
folder="/uploads/",
tags=["backend-upload"]
)
post = Post(
caption=caption,
url=response.url,
file_type="video" if file.content_type.startswith("video/") else "image",
file_name=response.name
)
session.add(post)
await session.commit()
await session.refresh(post)
return post
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
file.file.close()
Similarly we also create the fetch all post route and the delete post route.
Setting-up the authentication
The next very important aspect of any backend system is the authentication. Below is a detailed walkthrough of how I did it. First, we define the Users table in the db.py file. We set the user as a foreign key in the Post table, thus forming a one to many relation.
class User(Base, SQLAlchemyBaseUserTableUUID):
__tablename__ = "user"
posts = relationship("Post", back_populates="user")
class Post(Base):
__tablename__ = "posts"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
user_id = Column(UUID(as_uuid=True), ForeignKey("user.id"), nullable=False)
caption = Column(Text)
url = Column(String, nullable=False)
file_type = Column(String, nullable=False)
file_name = Column(String, nullable=False)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
user = relationship("User", back_populates="posts")
In the above code, the relationship function plays a very important role. Since the User is a foreign key in the Posts table, when we fetch a post showing the user information normally, it will only fetch the user ID which could be found in the Post table. What if we wanted more data of the user, we'd have to run separate sql queries manually to do so. But the relationships method does that automatically for us. Similarly, accessing the User information, we can also get the complete post information of the user.
async def get_user_db(session: AsyncSession = Depends(get_async_session)):
yield SQLAlchemyUserDatabase(session, User)
get_user_db() creates a FastAPI dependency that gives FastAPI Users a database interface (SQLAlchemyDatabase) connected to your User table through the current SQLAlchemy session, allowing the library to perform user-related database operations automatically
Next we create a users.py which consists of a UserManager, basically a class which consists of the user operations that need to be performed. This class consists of predefined operations from the FastAPI Users library which we can just use as a plug-n-play thing. Now most of the users.py file is boilerplate code, here's the code below:
import uuid
from typing import Optional
from fastapi import Depends, Request
from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin, models
from fastapi_users.authentication import AuthenticationBackend, BearerTransport, JWTStrategy
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase
from app.db import User, get_user_db
SECRET = "yourjwtsecret"
class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
reset_password_token_secret = SECRET
verification_token_secret = SECRET
async def on_after_register(self, user: User, request: Optional[Request] = None):
print(f"User {user.id} has registered.")
async def on_after_forgot_password(self, user: User, token: str, request: Optional[Request] = None):
print(f"User {user.id} has forgot their password. Reset token: {token}")
async def on_after_request_verify(self, user: User, token: str, request: Optional[Request] = None):
print(f"User {user.id} has requested verification. Verification token: {token}")
async def get_user_manager(user_db: SQLAlchemyUserDatabase = Depends(get_user_db)):
yield UserManager(user_db)
bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")
def get_jwt_strategy() -> JWTStrategy:
return JWTStrategy(secret=SECRET, lifetime_seconds=3600)
auth_backend = AuthenticationBackend(
name="jwt",
transport=bearer_transport,
get_strategy=get_jwt_strategy
)
fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])
current_active_user = fastapi_users.current_user(active=True)
Most of these functions like the on_after_register are basically predefined hooks. We can customize them by calling custom methods within those hooks.
One last change before the final integration is to build the schemas for user read, create and update. We'll just use the predefined Fast API schemas for now.
from pydantic import BaseModel
from fastapi_users import schemas
import uuid
class UserRead(schemas.BaseUser[uuid.UUID]):
pass
class UserCreate(schemas.BaseUserCreate):
pass
class UserUpdate(schemas.BaseUserUpdate):
pass
Next, we make the following changes to the app.py file:
from app.users import current_active_user, auth_backend, fastapi_users
from app.schemas import UserRead, UserCreate, UserUpdate
Importing the classes and methods.
app.include_router(fastapi_users.get_auth_router(auth_backend), prefix="/auth/jwt", tags=["auth"])
app.include_router(fastapi_users.get_register_router(UserRead, UserCreate), prefix="/auth", tags=["auth"])
app.include_router(fastapi_users.get_reset_password_router(), prefix="/auth", tags=["auth"])
app.include_router(fastapi_users.get_verify_router(UserRead), prefix="/auth", tags=["auth"])
app.include_router(fastapi_users.get_users_router(UserRead, UserUpdate), prefix="/users", tags=["users"])
This will add those routes to the FastAPI app.
Protecting the routes
Route protection means preventing unauthorized operations by restricting CRUD operations to only the active user instead of it being available to all the users. Below every route parameter, include the following line and it's done.
user: User = Depends(current_active_user),
That's done. FastAPI backend basics are complete.