Save User to Database – FastAPI - Beginner to Pro

📝 Saving a User to the Database

In this section, we will extend our create user endpoint to persist users in the database instead of just returning the model. We will also integrate dependency injection for database sessions.

Current Create User Endpoint

Currently, our create_user endpoint creates a user model but does not save it:

🗂️ routers/auth.py

@router.post("/auth/")
async def create_user(create_user_request: CreateUserRequest):
		......
		...

    return create_user_model

✅ Issue

The user exists only in memory.
No database transaction is executed.

We’ll need to add a database dependency and update the endpoint by adding a status code.

1️⃣ Adding Database Dependency

We want to reuse the pattern from our todos.py:

🗂️ routers/auth.py

from typing import Annotated
from sqlalchemy.orm import Session
from fastapi import Depends
from database import SessionLocal

{....../}

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

{......./}

This allows us to inject a database session into any endpoint.

2️⃣ Adding Status Code

🗂️ routers/auth.py

from starlette import status

@router.post("/auth/", status_code=status.HTTP_201_CREATED)
async def create_user(
    create_user_request: CreateUserRequest,
    db: db_dependency):

    # Create a new user model
    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        role=create_user_request.role,
        hashed_password=bcrypt_context.hash(create_user_request.password),
        is_active=True
    )

    # Save to database
    db.add(create_user_model)
    db.commit()

🗂️ routers/auth.py

from typing import Annotated
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from starlette import status
from database import SessionLocal
from models import Users
from passlib.context import CryptContext

router = APIRouter()

bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class CreateUserRequest(BaseModel):
    username: str
    email: str
    first_name: str
    last_name: str
    password: str
    role: str

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

@router.post("/auth/", status_code=status.HTTP_201_CREATED)
async def create_user(
    create_user_request: CreateUserRequest,
    db: db_dependency):

    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        role=create_user_request.role,
        hashed_password=bcrypt_context.hash(create_user_request.password),
        is_active=True
    )

    db.add(create_user_model)
    db.commit()

No Other Change

4️⃣ Testing User Creation

Start the server:

uvicorn main:app --reload

Open Swagger UI: http://localhost:8000/docs
Use the POST /auth/ endpoint and fill in a demo user.
Response:

Status code: 201 Created
Returns the created user (hashed password is stored, not plain text)

5️⃣ Verifying Data in Database

Open SQLite:

sqlite3 todosapp.db

Query users:

SELECT * FROM users;

You should see one user:

✅ Now the user is fully persisted in the database.

🗂️ routers/auth.py

from typing import Annotated
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from starlette import status
from database import SessionLocal
from models import Users
from passlib.context import CryptContext

router = APIRouter()

# Password hashing setup
bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# Pydantic schema for user creation
class CreateUserRequest(BaseModel):
    username: str
    email: str
    first_name: str
    last_name: str
    password: str
    role: str

# Database dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

# Create user endpoint
@router.post("/auth/", status_code=status.HTTP_201_CREATED)
async def create_user(
    create_user_request: CreateUserRequest,
    db: db_dependency):

    # Create a new user model
    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        role=create_user_request.role,
        hashed_password=bcrypt_context.hash(create_user_request.password),
        is_active=True
    )

    # Save to database
    db.add(create_user_model)
    db.commit()

No Other Change