Setting Up User Profiles in FastAPI

undraw svg icon

Welcome to Part 9 of Up and Running with FastAPI. If you missed part 8, you can find it here.

This series is focused on building a full-stack application with the FastAPI framework. The app allows users to post requests to have their residence cleaned, and other users can select a cleaning project for a given hourly rate.

Up And Running With FastAPI

In the previous post we implemented a proper login flow using FastAPI's built in OAuth2 system. On top of that, we built out dependencies that ensure users can access protected routes using JSON Web Tokens. With authentication out of the way, we can focus on user presence and how users interact with our application. We'll be building out user profiles and ownership, which is a non-trivial task. This post will be particularly long and most solutions will simply require us to bust out some SQL, so get ready.

Let's begin by giving users the ability to customize their profiles.

Creating User Profiles

We'll need to update our database to support profiles, as we'll want to give our users the ability to customize their online presence and interact with other user's profiles. Every user's id serves as a foreign key for items they own, and profiles are no different. So we'll need to account for that as well.

As with any modification to the database, we start with the migrations file.

Migrations

Before we make any adjustments, let's roll back our migrations.

docker ps
docker exec -it [CONTAINER_ID] bash
alembic downgrade base

Now go ahead and open up the file that looks like: db/migrations/versions/12345678654_create_main_tables.py.

"""create main tables
Revision ID: 12345678654
Revises:
Create Date: 2020-05-05 10:41:35.468471
"""
from typing import Tuple
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic
revision = "895636233437"
down_revision = None
branch_labels = None
depends_on = None
def create_updated_at_trigger() -> None:
op.execute(
"""
CREATE OR REPLACE FUNCTION update_updated_at_column()
RETURNS TRIGGER AS
$$
BEGIN
NEW.updated_at = now();
RETURN NEW;
END;
$$ language 'plpgsql';
"""
)
def timestamps() -> Tuple[sa.Column, sa.Column]:
return (
sa.Column("created_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.TIMESTAMP(timezone=True), server_default=sa.func.now(), nullable=False),
)
def create_cleanings_table() -> None:
op.create_table(
"cleanings",
sa.Column("id", sa.Integer, primary_key=True),
sa.Column("name", sa.Text, nullable=False, index=True),
sa.Column("description", sa.Text, nullable=True),
sa.Column("cleaning_type", sa.Text, nullable=False, server_default="spot_clean"),
sa.Column("price", sa.Numeric(10, 2), nullable=False),
*timestamps(),
)
op.execute(
"""
CREATE TRIGGER update_cleanings_modtime
BEFORE UPDATE
ON cleanings
FOR EACH ROW
EXECUTE PROCEDURE update_updated_at_column();
"""
)
def create_users_table() -> None:
op.create_table(
"users",
sa.Column("id", sa.Integer, primary_key=True),
sa.Column("username", sa.Text, unique=True, nullable=False, index=True),
sa.Column("email", sa.Text, unique=True, nullable=False, index=True),
sa.Column("email_verified", sa.Boolean, nullable=False, server_default="False"),
sa.Column("salt", sa.Text, nullable=False),
sa.Column("password", sa.Text, nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=False),
sa.Column("is_superuser", sa.Boolean(), nullable=False),
*timestamps(),
)
op.execute(
"""
CREATE TRIGGER update_user_modtime
BEFORE UPDATE
ON users
FOR EACH ROW
EXECUTE PROCEDURE update_updated_at_column();
"""
)
def create_profiles_table() -> None:
op.create_table(
"profiles",
sa.Column("id", sa.Integer, primary_key=True),
sa.Column("full_name", sa.Text, nullable=True),
sa.Column("phone_number", sa.Text, nullable=True),
sa.Column("bio", sa.Text, nullable=True, server_default=""),
sa.Column("image", sa.Text, nullable=True),
sa.Column("user_id", sa.Integer, sa.ForeignKey("users.id", ondelete="CASCADE")),
*timestamps(),
)
op.execute(
"""
CREATE TRIGGER update_profiles_modtime
BEFORE UPDATE
ON profiles
FOR EACH ROW
EXECUTE PROCEDURE update_updated_at_column();
"""
)
def upgrade() -> None:
create_updated_at_trigger()
create_cleanings_table()
create_users_table()
create_profiles_table()
def downgrade() -> None:
op.drop_table("profiles")
op.drop_table("users")
op.drop_table("cleanings")
op.execute("DROP FUNCTION update_updated_at_column")

A couple things to note here. We're adding a profiles table to our database that stores supplementary information about a user. We're using the SQLAlchemy's sa.ForeignKey table constraint to specify that each record in the profiles table belongs to a record in the users table.

While it's often convenient to keep all user-related information in a single table, we aren't taking that approach here. Instead, we'll keep authentication information in the users table and personal information in the profiles table. When we want to get both, we'll simply join the tables in our SQL query.

In our case, the flexibility of this approach outweighs the cost of joining the tables whenever a user is queried. We're also able to add an arbitrary number of columns to the profile model and attach an unlimited number of profile types to a single user.

At the end of the file, we've adjusted the upgrade and downgrade functions for our migrations runner.

With that out the way, it's time to migrate the database by entering the container like before (unless we're already there) and running the alembic upgrade head command.

docker ps
docker exec -it [CONTAINER_ID] bash
alembic upgrade head

Profile Models

Let's go ahead and create models for our user profiles.

Create a new file in the models directory called profile.py.

touch backend/app/models.profile.py

And inside that file, add the following:

models/profile.py
from typing import Optional
from pydantic import EmailStr, constr, HttpUrl
from app.models.core import DateTimeModelMixin, IDModelMixin, CoreModel
class ProfileBase(CoreModel):
full_name: Optional[str]
phone_number: Optional[constr(regex="^\d{1,3}-\d{1,3}?-\d{1,4}?$")]
bio: Optional[str]
image: Optional[HttpUrl]
class ProfileCreate(ProfileBase):
"""
The only field required to create a profile is the users id
"""
user_id: int
class ProfileUpdate(ProfileBase):
"""
Allow users to update any or no fields, as long as it's not user_id
"""
pass
class ProfileInDB(IDModelMixin, DateTimeModelMixin, ProfileBase):
user_id: int
username: Optional[str]
email: Optional[EmailStr]
class ProfilePublic(ProfileInDB):
pass

Nothing crazy going on here. The profile is standard, so most of the work is done by inheriting from our base model and mixins. Though our profiles table doesn't have a username field or an email field, we still add them to the ProfileInDB model. The ProfilePublic model inherits them as well. Depending on the situation, this may be useful for displaying user profiles in our UI.

One additional interesting thing we've done is to add a simple regex validation to phone numbers that checks for a string matching the pattern: 555-888-9500. We've also specified that the image must by an http url - validated for us by pydantic.

Testing User Profiles

Let's make some tests. We want tests that ensure a profile is created for a user when they register, that users can see other users' profiles when they're authenticated, and that users can update their own profile. In a future post, we'll add a social component to our application and test that as well.

We'll do this in pieces, taking it one step at a time.

Start, by creating a new file called test_profiles.py.

touch backend/tests/test_profiles.py

And add the following to it.

tests/test_profiles.py
import pytest
from fastapi import FastAPI, status
from httpx import AsyncClient
from app.models.user import UserInDB
pytestmark = pytest.mark.asyncio
class TestProfilesRoutes:
"""
Ensure that no api route returns a 404
"""
async def test_routes_exist(self, app: FastAPI, client: AsyncClient, test_user: UserInDB) -> None:
# Get profile by username
res = await client.get(app.url_path_for("profiles:get-profile-by-username", username=test_user.username))
assert res.status_code != status.HTTP_404_NOT_FOUND
# Update own profile
res = await client.put(app.url_path_for("profiles:update-own-profile"), json={"profile_udpate": {}})
assert res.status_code != status.HTTP_404_NOT_FOUND

Baby steps here, as we're only checking to see if 2 routes exists: one to fetch a profile by a user's username and one to update a user's own profile. We've also started using from fastapi import status for our status codes. No specific reason for doing it this way. We simply take this approach because it requires fewer lines than importing status codes from starlette directly.

Run the tests and watch them fail.

docker ps
docker exec -it [CONTAINER_ID] bash
pytest tests/test_profiles.py -v -p no:warnings

Making these two tests pass is easy enough. Let's create a new route file in the api/routes/ directory.

touch backend/app/api/routes/profiles.py

And we can start adding to it like so:

api/routes/profiles.py
from fastapi import APIRouter, Path, Body
from app.models.profile import ProfileUpdate, ProfilePublic
router = APIRouter()
@router.get("/{username}/", response_model=ProfilePublic, name="profiles:get-profile-by-username")
async def get_profile_by_username(
*, username: str = Path(..., min_length=3, regex="[a-zA-Z0-9_-]+$"),
) -> ProfilePublic:
return None
@router.put("/me/", response_model=ProfilePublic, name="profiles:update-own-profile")
async def update_own_profile(profile_update: ProfileUpdate = Body(..., embed=True)) -> ProfilePublic:
return None

We've defined a GET route and a PUT route for fetching and updating profiles, respectively. No logic here yet, as they both simply return None. The only thing to pay attention to is that we're validating the username in the same way as in our UserCreate and UserUpdate models. It must be at least 3 characters long and consist of only letters, numbers, underscores and dashes.

We'll need to register this new router with our api router so open up the api/routes/__init__.py file.

api/routes/__init__.py
from fastapi import APIRouter
from app.api.routes.cleanings import router as cleanings_router
from app.api.routes.users import router as users_router
from app.api.routes.profiles import router as profiles_router
router = APIRouter()
router.include_router(cleanings_router, prefix="/cleanings", tags=["cleanings"])
router.include_router(users_router, prefix="/users", tags=["users"])
router.include_router(profiles_router, prefix="/profiles", tags=["profiles"])

Same as before, we attach our profiles router to the api router under the /profiles namespace.

Now when we run our tests again they should pass.

Let's move on and add the next test class to our test_profiles.py file.

tests/test_profiles.py
# ... other code
from app.models.user import UserInDB, UserPublic
from app.models.profile import ProfileInDB
from app.db.repositories.profiles import ProfilesRepository
# ...other code
class TestProfileCreate:
async def test_profile_created_for_new_users(self, app: FastAPI, client: AsyncClient, db: Database) -> None:
profile_repo = ProfilesRepository(db)
new_user = {"email": "dwayne@johnson.io", "username": "therock", "password": "dwaynetherockjohnson"}
res = await client.post(app.url_path_for("users:register-new-user"), json={"new_user": new_user})
assert res.status_code == status.HTTP_201_CREATED
created_user = UserPublic(**res.json())
user_profile = await profile_repo.get_profile_by_user_id(user_id=created_user.id)
assert user_profile is not None
assert isinstance(user_profile, ProfileInDB)

As soon as we try to run our tests, we'll get an import error. We don't have a ProfilesRepository yet, so let's make one.

touch backend/app/db/repositories/profiles.py

And let's build it out with two new fancy methods.

db/repositories/profiles.py
from app.db.repositories.base import BaseRepository
from app.models.profile import ProfileCreate, ProfileUpdate, ProfileInDB
CREATE_PROFILE_FOR_USER_QUERY = """
INSERT INTO profiles (full_name, phone_number, bio, image, user_id)
VALUES (:full_name, :phone_number, :bio, :image, :user_id)
RETURNING id, full_name, phone_number, bio, image, user_id, created_at, updated_at;
"""
GET_PROFILE_BY_USER_ID_QUERY = """
SELECT id, full_name, phone_number, bio, image, user_id, created_at, updated_at
FROM profiles
WHERE user_id = :user_id;
"""
class ProfilesRepository(BaseRepository):
async def create_profile_for_user(self, *, profile_create: ProfileCreate) -> ProfileInDB:
created_profile = await self.db.fetch_one(query=CREATE_PROFILE_FOR_USER_QUERY, values=profile_create.dict())
return created_profile
async def get_profile_by_user_id(self, *, user_id: int) -> ProfileInDB:
profile_record = await self.db.fetch_one(query=GET_PROFILE_BY_USER_ID_QUERY, values={"user_id": user_id})
if not profile_record:
return None
return ProfileInDB(**profile_record)

Our new ProfilesRepository is now ready for use. It can create profiles for new users and fetch a profile when provided the user_id. If we run our tests now, we should see that our latest test is failing. When we attempt to fetch the newly created user's profile, we get None. Let's make sure that when a new user is created, our UsersRepository also creates a profile for that user.

Open up the db/repositories/users.py file and update it like so:

db/repositories/users.py
# ...other code
from app.db.repositories.profiles import ProfilesRepository
from app.models.profile import ProfileCreate
# ...other code
class UsersRepository(BaseRepository):
def __init__(self, db: Database) -> None:
super().__init__(db)
self.auth_service = auth_service
self.profile_repo = ProfilesRepository(db)
# ...other code
async def register_new_user(self, *, new_user: UserCreate) -> UserInDB:
# make sure email isn't already taken
if await self.get_user_by_email(email=new_user.email):
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST,
detail="That email is already taken. Login with that email or register with another one."
)
# make sure username isn't already taken
if await self.get_user_by_username(username=new_user.username):
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST,
detail="That username is already taken. Please try another one."
)
user_password_update = self.auth_service.create_salt_and_hashed_password(plaintext_password=new_user.password)
new_user_params = new_user.copy(update=user_password_update.dict())
created_user = await self.db.fetch_one(query=REGISTER_NEW_USER_QUERY, values=new_user_params.dict())
await self.profile_repo.create_profile_for_user(profile_create=ProfileCreate(user_id=created_user.id))
return UserInDB(**created_user)
# ...other code

This is a useful pattern that we'll take advantage of regularly. By adding the ProfilesRepository as a sub-repo of the UsersRepository, we can insert any profile-related logic directly into our user-related logic. And we do just that here. Once a user registers with our application, we take the newly created user's id and use it to add an empty profile to our database. If we want to allow users to sign up with additional information, we can pass that along here as well.

Run the tests again and they should all pass.

Fetching and Updating Profiles

Let's flesh out those two empty routes we created earlier, starting with some tests.

In the conftest.py file, add a new fixture.

tests/conftest.py
# ...other code
@pytest.fixture
async def test_user2(db: Database) -> UserInDB:
new_user = UserCreate(
email="serena@williams.io",
username="serenawilliams",
password="tennistwins",
)
user_repo = UsersRepository(db)
existing_user = await user_repo.get_user_by_email(email=email)
if existing_user:
return existing_user
return await user_repo.register_new_user(new_user=new_user)

Now, let's use it in our test_profiles.py file.

tests/test_profiles.py
# ...other code
class TestProfileView:
async def test_authenticated_user_can_view_other_users_profile(
self, app: FastAPI, authorized_client: AsyncClient, test_user: UserInDB, test_user2: UserInDB
) -> None:
res = await authorized_client.get(
app.url_path_for("profiles:get-profile-by-username", username=test_user2.username)
)
assert res.status_code == status.HTTP_200_OK
profile = ProfilePublic(**res.json())
assert profile.username == test_user2.username
async def test_unregistered_users_cannot_access_other_users_profile(
self, app: FastAPI, client: AsyncClient, test_user2: UserInDB
) -> None:
res = await authorized_client.get(
app.url_path_for("profiles:get-profile-by-username", username=test_user2.username)
)
assert res.status_code == status.HTTP_401_UNAUTHORIZED
async def test_no_profile_is_returned_when_username_matches_no_user(
self, app: FastAPI, authorized_client: AsyncClient
) -> None:
res = await authorized_client.get(
app.url_path_for("profiles:get-profile-by-username", username="username_doesnt_match")
)
assert res.status_code == status.HTTP_404_NOT_FOUND

In the first test, we check to see if test_user can access the profile of test_user2. Since our authorized_client uses the JWT token for test_user, this is relatively straightforward to implement. In the second test, we attempt to do the same thing, except with an unauthorized client. We expect to see it fail. Our third test simply ensures that a 404 is returned when the username has no corresponding profile.

Run the tests and watch them fail.

Let's start in the api/routes/profiles.py file. Here's the updated file.

api/routes/profiles.py
from fastapi import Depends, APIRouter, HTTPException, Path, Body, status
from app.api.dependencies.auth import get_current_active_user
from app.api.dependencies.database import get_repository
from app.models.user import UserCreate, UserUpdate, UserInDB, UserPublic
from app.models.profile import ProfileUpdate, ProfilePublic
from app.db.repositories.profiles import ProfilesRepository
router = APIRouter()
@router.get("/{username}/", response_model=ProfilePublic, name="profiles:get-profile-by-username")
async def get_profile_by_username(
username: str = Path(..., min_length=3, regex="[a-zA-Z0-9_-]+$"),
current_user: UserInDB = Depends(get_current_active_user),
profile_repo: ProfilesRepository = Depends(get_repository(ProfilesRepository)),
) -> ProfilePublic:
profile = await profile_repo.get_profile_by_username(username=username)
if not profile:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No profile found with that username.")
return profile
@router.put("/me/", response_model=ProfilePublic, name="profiles:update-own-profile")
async def update_own_profile(profile_update: ProfileUpdate = Body(..., embed=True)) -> ProfilePublic:
return None

Run the tests again and the last one should pass. Simply by including the get_current_active_user dependency, we protect this route from unauthenticated requests. The other two tests are failing because we haven't implemented the get_profile_by_username method on our ProfilesRepository.

This will be mostly an exercise in SQL, so let's get to it.

db/repositories/profiles.py
from app.models.user import UserInDB
# ...other code
GET_PROFILE_BY_USERNAME_QUERY = """
SELECT p.id,
u.email AS email,
u.username AS username,
full_name,
phone_number,
bio,
image,
user_id,
p.created_at,
p.updated_at
FROM profiles p
INNER JOIN users u
ON p.user_id = u.id
WHERE user_id = (SELECT id FROM users WHERE username = :username);
"""
class ProfilesRepository(BaseRepository):
# ...other code
async def get_profile_by_username(self, *, username: str) -> ProfileInDB:
profile_record = await self.db.fetch_one(query=GET_PROFILE_BY_USERNAME_QUERY, values={"username": username})
if not profile_record:
return None
return ProfileInDB(**profile_record)

Before we get to the SQL, let's talk about what we're trying to accomplish. We want to take in a username and check in our database for any user with that username. If we find that username, we want to grab their email and username. Then we want to attach it to the profile associated with that user and return the ProfileInDB model with all of the attributes.

To make that all happen, we join the profiles and users table together for a user that matches the sub-query:

SELECT id FROM users WHERE username = :username;

We then only select the username and email from the users table, while selecting all fields from the profiles table.

Run the tests again and they should all pass.

Attaching Profiles to UserPublic Models

Let's refactor our application models a bit. We're going to attach public user profiles to user models that are returned by our user routes.

Start with the models/user.py file and update it like so:

models/user.py
# ...other code
from app.models.profile import ProfilePublic
# ...other code
class UserInDB(IDModelMixin, DateTimeModelMixin, UserBase):
"""
Add in user's password and salt. Allow optional user profile
"""
password: constr(min_length=7)
salt: str
profile: Optional[ProfilePublic]
class UserPublic(IDModelMixin, DateTimeModelMixin, UserBase):
access_token: Optional[AccessToken]
profile: Optional[ProfilePublic]

Now we have the ability to attach a user profile to the user models. In the UsersRepository add a new method that makes it easy to do populate the user with their profile.

Here's the whole file in its entirety.

db/repositories/users.py
from typing import Optional
from pydantic import EmailStr
from fastapi import HTTPException, status
from databases import Database
from app.db.repositories.base import BaseRepository
from app.db.repositories.profiles import ProfilesRepository
from app.models.profile import ProfileCreate, ProfilePublic
from app.models.user import UserCreate, UserUpdate, UserInDB
from app.services import auth_service
GET_USER_BY_EMAIL_QUERY = """
SELECT id, username, email, email_verified, password, salt, is_active, is_superuser, created_at, updated_at
FROM users
WHERE email = :email;
"""
GET_USER_BY_USERNAME_QUERY = """
SELECT id, username, email, email_verified, password, salt, is_active, is_superuser, created_at, updated_at
FROM users
WHERE username = :username;
"""
REGISTER_NEW_USER_QUERY = """
INSERT INTO users (username, email, email_verified, password, salt, is_active, is_superuser)
VALUES (:username, :email, :email_verified, :password, :salt, :is_active, :is_superuser)
RETURNING id, username, email, email_verified, password, salt, is_active, is_superuser, created_at, updated_at;
"""
class UsersRepository(BaseRepository):
def __init__(self, db: Database) -> None:
super().__init__(db)
self.auth_service = auth_service
self.profile_repo = ProfilesRepository(db)
async def get_user_by_email(self, *, email: EmailStr) -> UserInDB:
user_record = await self.db.fetch_one(query=GET_USER_BY_EMAIL_QUERY, values={"email": email})
if not user_record:
return None
return await self.populate_user(user=UserInDB(**user_record))
async def get_user_by_username(self, *, username: str) -> UserInDB:
user_record = await self.db.fetch_one(query=GET_USER_BY_USERNAME_QUERY, values={"username": username})
if not user_record:
return None
return await self.populate_user(user=UserInDB(**user_record))
async def register_new_user(self, *, new_user: UserCreate) -> UserInDB:
if await self.get_user_by_email(email=new_user.email):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="That email is already taken. Login with that email or register with another one.",
)
if await self.get_user_by_username(username=new_user.username):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="That username is already taken. Please try another one.",
)
user_password_update = self.auth_service.create_salt_and_hashed_password(plaintext_password=new_user.password)
new_user_params = new_user.copy(update=user_password_update.dict())
created_user = await self.db.fetch_one(query=REGISTER_NEW_USER_QUERY, values=new_user_params.dict())
# create profile for new user
profile = await self.profile_repo.create_profile_for_user(
profile_create=ProfileCreate(user_id=created_user["id"])
)
return UserInDB(**created_user, profile=profile)
async def authenticate_user(self, *, email: EmailStr, password: str) -> Optional[UserInDB]:
user = await self.get_user_by_email(email=email)
if not user:
return None
verified, updated_password_hash = self.auth_service.verify_password(
password=password, salt=user.salt, hashed_pw=user.password
)
if not verified:
return None
return user
async def populate_user(self, *, user: UserInDB) -> UserInDB:
user_profile = await self.profile_repo.get_profile_by_user_id(user_id=user.id)
user.profile = ProfilePublic(**user_profile.dict())
return user

And voila! Users should now have their profiles attached when they are returned from our API. Try it out in the interactive docs at localhost:8000/docs.

Updating Profiles

Now let's make sure users can update their own profiles.

Create a new test class in test_profiles.py.

tests/test_profiles.py
# ...other code
class TestProfileManagement:
@pytest.mark.parametrize(
"attr, value",
(
("full_name", "Lebron James"),
("phone_number", "555-333-1000"),
("bio", "This is a test bio"),
("image", "http://testimages.com/testimage"),
),
)
async def test_user_can_update_own_profile(
self, app: FastAPI, authorized_client: AsyncClient, test_user: UserInDB, attr: str, value: str,
) -> None:
assert getattr(test_user.profile, attr) != value
res = await authorized_client.put(
app.url_path_for("profiles:update-own-profile"), json={"profile_update": {attr: value}},
)
assert res.status_code == status.HTTP_200_OK
profile = ProfilePublic(**res.json())
assert getattr(profile, attr) == value
@pytest.mark.parametrize(
"attr, value, status_code",
(
("full_name", [], 422),
("phone_number", "5555-3335-1000555", 422),
("phone_number", "wrong number", 422),
("bio", {}, 422),
("image", "./image-string.png", 422),
("image", 5, 422),
),
)
async def test_user_recieves_error_for_invalid_update_params(
self,
app: FastAPI,
authorized_client: AsyncClient,
test_user: UserInDB,
attr: str,
value: str,
status_code: int,
) -> None:
res = await authorized_client.put(
app.url_path_for("profiles:update-own-profile"), json={"profile_update": {attr: value}},
)
assert res.status_code == status_code

Run the tests and watch them fail.

Open up the api/routes/profiles.py file and update the PUT route like so:

api/routes/profiles.py
# ...other code
@router.put("/me/", response_model=ProfilePublic, name="profiles:update-own-profile")
async def update_own_profile(
profile_update: ProfileUpdate = Body(..., embed=True),
current_user: UserInDB = Depends(get_current_active_user),
profile_repo: ProfilesRepository = Depends(get_repository(ProfilesRepository)),
) -> ProfilePublic:
updated_profile = await profile_repo.update_profile(profile_update=profile_update, requesting_user=current_user)
return updated_profile

We're simply calling the currently non-existent update_profile method and passing along whatever updates are needed along with the user that is being updated.

Now on to the repo.

db/repositories/profiles.py
# ...other code
UPDATE_PROFILE_QUERY = """
UPDATE profiles
SET full_name = :full_name,
phone_number = :phone_number,
bio = :bio,
image = :image
WHERE user_id = :user_id
RETURNING id, full_name, phone_number, bio, image, user_id, created_at, updated_at;
"""
class ProfilesRepository(BaseRepository):
# ...other code
async def update_profile(self, *, profile_update: ProfileUpdate, requesting_user: UserInDB) -> ProfileInDB:
profile = await self.get_profile_by_user_id(user_id=requesting_user.id)
update_params = profile.copy(update=profile_update.dict(exclude_unset=True))
updated_profile = await self.db.fetch_one(
query=UPDATE_PROFILE_QUERY,
values=update_params.dict(exclude={"id", "created_at", "updated_at", "username", "email"}),
)
return ProfileInDB(**updated_profile)

Run the tests again and this time they should pass.

Wrapping Up and Resources

This was the longest post so far, so readers who feel exhausted by this point should not feel bad. Our API is really taking shape, and we're ready to put the finishing touches on an MVP. In the next post, we'll ensure users own cleanings they create and refactor all of the associated routes and repository methods.

  • SQL Joins lesson from SQL Bolt
  • Alembic migration docs
  • SQLAlchemy constraints docs