FastAPI を使ってWEBアプリを作ってみる その9
前回の投稿ではFastAPIに組み込まれたOAuth2システムを使ってログインフローを実装し、JSON Web Tokensを使用してユーザーが保護されたルートにアクセスできるように依存関係を構築しました。
今回の投稿ではユーザーが自分のプロフィールをカスタマイズできるようにすることから始めます。
過去の投稿はこちらから辿ることができます。
| FastAPI を使ってWEBアプリを作ってみる その1 | FastAPIとDockerでHelloWorld |
| FastAPI を使ってWEBアプリを作ってみる その2 | AlembicとPostgreSQLでDB Migrate |
| FastAPI を使ってWEBアプリを作ってみる その3 | APIエンドポイントをPostgreSQLに接続 |
| FastAPI を使ってWEBアプリを作ってみる その4 | PytestとDockerでテスト構築 |
| FastAPI を使ってWEBアプリを作ってみる その5 | RESTAPIエンドポイントを実装 |
| FastAPI を使ってWEBアプリを作ってみる その6 | ユーザーモデルの作成 |
| FastAPI を使ってWEBアプリを作ってみる その7 | パスワードの暗号化とJWT認証 |
| FastAPI を使ってWEBアプリを作ってみる その8 | 認証機能のDependencies化 |
| FastAPI を使ってWEBアプリを作ってみる その9 | 今ここ |
ユーザープロファイルの作成
ユーザーが自分のプロファイルをカスタマイズ可能にするためにデータベースを更新してプロファイルテーブルを作成します。
まずmigrationsファイルを作成します。
マイグレーション
更新する前にマイグレーションをロールバックしましょう:
$ alembic downgrade base"""create_first_tables
Revision ID: foobar
Revises:
Create Date: 2021-01-22 18:27:50.234519
"""
...省略
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_hedgehogs_table()
create_users_table()
create_profiles_table() # 追加
def downgrade() -> None:
op.drop_table("profiles") # 追加
op.drop_table("users")
op.drop_table("hedgehogs")
op.execute("DROP FUNCTION update_updated_at_column")ユーザーに関する補足的な情報を保存するプロファイルテーブルをデータベースに追加しています。SQLAlchemyのsa.ForeignKeyテーブル制約を使って、profilesテーブルの各レコードがusersテーブルのレコードに属するように指定しています。
すべてのユーザー関連情報をひとつのテーブルにまとめておくと便利なことが多いのですが、ここではそのような方法はとりません。
代わりに認証情報をusersテーブルに、個人情報をprofilesテーブルに格納し両方を取得したい場合はSQLクエリでテーブルを結合し取得します。
今回のケースではユーザーが検索されるたびにテーブルを結合するコストを柔軟性が上回る想定です。
この変更をデータベースにマイグレートしましょう:
$ alembic upgrade headプロファイルモデルの作成
早速ユーザープロファイルのモデルを作ってみましょう。
modelsディレクトリに profile.py というファイルを作成してください:
$ touch app/models/profile.pyfrom typing import Optional
from app.models.core import CoreModel, DateTimeModelMixin, IDModelMixin
from pydantic import EmailStr, HttpUrl
class ProfileBase(CoreModel):
full_name: Optional[str]
phone_number: Optional[str]
bio: Optional[str]
image: Optional[HttpUrl]
class ProfileCreate(ProfileBase):
user_id: int
class ProfileUpdate(ProfileBase):
pass
class ProfileInDB(IDModelMixin, DateTimeModelMixin, ProfileBase):
user_id: int
username: Optional[str]
email: Optional[EmailStr]
class ProfilePublic(ProfileInDB):
passほとんどのモデルはベースモデルとMixInを継承して定義しています。
先ほど定義したプロファイルテーブルには username フィールドも email フィールドもありませんがProfileInDB モデルにそれらを追加しています。
ProfilePublic モデルも同様にそれらを継承します。
こうしておくと状況に応じてユーザーのプロフィールをUIに表示する際にフロント側のコードがシンプルになり便利です。
またimageには HttpUrl 型を指定していますがこれはpydanticで検証することができます。
ユーザプロファイルのテスト作成
それではいつも通り最初にテストを作ります。
サインアップ時にユーザープロフィールが作成されること、認証時にユーザーが他のユーザーのプロフィールを見ることができること、ユーザーが自分のプロフィールを更新できることを確認するテストが必要です。
今後の投稿では、SNSとの連携コンポーネントを追加してそのテストも作成する予定です。
まず、test_profiles.pyという新しいファイルを作成します:
$ touch tests/test_profiles.pyimport pytest
from app.models.user import UserInDB
from databases import Database
from fastapi import FastAPI, status
from httpx import AsyncClient
pytestmark = pytest.mark.asyncio
class TestProfilesRoutes:
async def test_routes_exist(
self, app: FastAPI, client: AsyncClient, test_user: UserInDB
) -> None:
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
res = await client.put(
app.url_path_for("profiles:update-own-profile"), json={"profile_update": {}}
)
assert res.status_code != status.HTTP_404_NOT_FOUND
この段階では2つのルートが存在するかどうかをチェックしているだけです。
1つはユーザーのユーザー名でプロファイルを取得するもの、もう1つはユーザー自身のプロファイルを更新するものです。
テストを実行して失敗するところを見てみましょう:
$ /backend # pytest tests/test_profiles.py
================= test session starts ==================
platform linux -- Python 3.9.1, pytest-6.2.2, py-1.10.0, pluggy-0.13.1
rootdir: /backend
plugins: cov-2.11.1, asyncio-0.14.0
collected 1 item
tests/test_profiles.py F [100%]
======================= FAILURES =======================
_________ TestProfilesRoutes.test_routes_exist _________
...省略安定のNot Existが返ってきたところでこのルートを作成します:
$ touch app/api/routes/profiles.pyfrom app.models.profile import ProfilePublic, ProfileUpdate
from fastapi import APIRouter, Body, Path
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
プロファイルを取得するための GET ルートと更新するための PUT ルートをそれぞれ定義しました。
どちらも None を返すだけでまだロジックはありません。
唯一注意しなければならないのは、UserCreateやUserUpdateモデルと同じ方法でユーザー名を検証していることです。
ユーザ名は少なくとも3文字以上で、文字、数字、アンダースコア、ダッシュのみで構成されていなければなりません。
この新しいルーターを登録しましょう:
from app.api.routes.hedgehogs import router as hedgehogs_router
from app.api.routes.profiles import router as profiles_router # 追加
from app.api.routes.users import router as users_router
from fastapi import APIRouter
router = APIRouter()
router.include_router(hedgehogs_router, prefix="/hedgehogs", tags=["hedgehogs"])
router.include_router(users_router, prefix="/users", tags=["users"])
router.include_router(profiles_router, prefix="/profiles", tags=["profiles"]) # 追加
これでテストを再度実行するとパスするはずです。
結果を確認できたら引き続きテストケースを追加していきます:
...省略
from app.db.repositories.profiles import ProfilesRepository
from app.models.profile import ProfileInDB, ProfilePublic
from app.models.user import UserInDB, UserPublic
...省略
class TestProfileCreate:
async def test_profile_created_for_new_users(
self, app: FastAPI, client: AsyncClient, db: Database
) -> None:
profiles_repo = ProfilesRepository(db)
new_user = {
"email": "nmomos@mail.com",
"username": "nmomoishedgehog",
"password": "nmomosissocute",
}
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 profiles_repo.get_profile_by_user_id(
user_id=created_user.id
)
assert user_profile is not None
assert isinstance(user_profile, ProfileInDB)
テストを実行するとすぐにインポートエラーが発生します。
ProfilesRepositoryがまだないので作りましょう:
$ touch app/db/repositories/profiles.pyfrom app.db.repositories.base import BaseRepository
from app.models.profile import ProfileCreate, ProfileInDB, ProfileUpdate
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)
新しいユーザーのプロファイルを作成したり、user_idを指定してプロファイルを取得したりすることができるリポジトリを作成しました。
ここでテストを実行してみると失敗していることがわかります。
新しく作成されたユーザーのプロファイルを取得しようとするとNoneが返されているためです。
これを修正し新しいユーザが作成されたときにUsersRepositoryがそのユーザのプロファイルも一緒に作成するよう更新します:
...省略
from app.db.repositories.base import BaseRepository
from app.db.repositories.profiles import ProfilesRepository
from app.models.profile import ProfileCreate
...省略
class UsersRepository(BaseRepository):
def __init__(self, db: Database) -> None:
super().__init__(db)
self.auth_service = auth_service
self.profiles_repo = ProfilesRepository(db) # 追加
...省略
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="このメールアドレスはすでに登録されています"
)
if await self.get_user_by_username(username=new_user.username):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="このユーザ名はすでに登録されています"
)
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.profiles_repo.create_profile_for_user(
profile_create=ProfileCreate(user_id=created_user["id"])
)
# ここまで
return UserInDB(**created_user)
...省略
これは定期的に利用する便利なパターンです。
ProfilesRepositoryをUsersRepositoryのサブリポジトリとして追加することで、プロファイル関連のロジックをユーザー関連のロジックに直接挿入することができます。
ユーザーがアプリケーションに登録すると新しく作成されたユーザーのIDを取得し、それを使ってデータベースに空のプロファイルを追加します。
ユーザーが追加の情報を入力して登録できるようにしたい場合はもこの方法でここで実装することができます。
これで test_profile_created_for_new_users テストを実行すると合格しているはずです。
プロファイルの取得と更新
先ほど作成した2つのルートにいくつかテストを追加し具体的な認証系のエラートラップが適切に返されるか検証しておきます。
まずは conftest.py ファイルに新しいフィクスチャを追加します:
...省略
@pytest.fixture
async def test_user2(db: Database) -> UserInDB:
new_user = UserCreate(
email="nmomos2@mail.com",
username="nmomoishedgehog2",
password="nmomosissocute2",
)
user_repo = UsersRepository(db)
existing_user = await user_repo.get_user_by_email(email=new_user.email)
if existing_user:
return existing_user
return await user_repo.register_new_user(new_user=new_user)
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 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
最初のテストでは、test_userがtest_user2のプロファイルにアクセスできるかどうかを確認します。authorized_clientはtest_userのJWTトークンを使用しているのでこれは比較的簡単に実装できます。
2つ目のテストでは、未承認のクライアントを使って同じことを試みます。
このテストは失敗することが予想されます。
3つ目のテストでは、ユーザ名に対応するプロファイルがない場合に404が返されることを確認します。
テストを実行して失敗させます。
エラーが示す通りに準備を進めましょう、api/routes/profiles.pyファイルをリファクタします:
from app.api.dependencies.auth import get_current_active_user
from app.api.dependencies.database import get_repository
from app.db.repositories.profiles import ProfilesRepository
from app.models.profile import ProfilePublic, ProfileUpdate
from app.models.user import UserCreate, UserInDB, UserPublic, UserUpdate
from fastapi import APIRouter, Body, Depends, HTTPException, Path, status
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),
profiles_repo: ProfilesRepository = Depends(get_repository(ProfilesRepository)),
) -> ProfilePublic:
profile = await profiles_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
テストを再度実行すると ProfilesRepository に get_profile_by_username メソッドを実装していないことに起因するエラーが返ってきます。
これに対応します:
...省略
from app.models.user import UserInDB
...省略
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):
...省略
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 profile_record:
return ProfileInDB(**profile_record)
ユーザー名を入力しそのユーザー名を持つユーザーがいないかデータベースをチェックします。
見つかったらEメールとユーザー名を取得します。
そして、そのユーザーに関連付けられたプロファイルにアタッチしすべての属性を持つProfileInDBモデルを返します。
users テーブルからは username と email のみを選択し、profiles テーブルからはすべてのフィールドを選択します。
テストを再度実行すると、すべてのテストがパスするはずです。
ユーザーモデルへプロファイルをアタッチ
アプリケーションモデルを少しリファクタリしユーザープロファイルをユーザールートから返されるユーザーモデルにアタッチしましょう。
models/user.py を更新します:
...省略
from app.models.profile import ProfilePublic
...省略
class UserPublic(IDModelMixin, DateTimeModelMixin, UserBase):
access_token: Optional[AccessToken]
profile: Optional[ProfilePublic]これでUserPublicモデルにユーザープロファイルを添付できるようになりました。
とても簡単な小さな変更ですがこれを機能させるにはいくつかの追加でコードを変更する必要があります。
UsersRepositoryに新しいメソッドを追加してユーザーのプロフィールを簡単に入力できるようにします:
...省略
from app.models.profile import ProfileCreate, ProfilePublic # 追加
...省略
class UsersRepository(BaseRepository):
...省略
# 更新ここから
async def get_user_by_email(
self, *, email: EmailStr, populate: bool = True
) -> UserInDB:
user_record = await self.db.fetch_one(
query=GET_USER_BY_EMAIL_QUERY, values={"email": email}
)
if user_record:
user = UserInDB(**user_record)
if populate:
return await self.populate_user(user=user)
return user
async def get_user_by_username(
self, *, username: str, populate: bool = True
) -> UserInDB:
user_record = await self.db.fetch_one(
query=GET_USER_BY_USERNAME_QUERY, values={"username": username}
)
if user_record:
user = UserInDB(**user_record)
if populate:
return await self.populate_user(user=user)
return user
#ここまで
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="このメールアドレスはすでに登録されています"
)
if await self.get_user_by_username(username=new_user.username):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="このユーザ名はすでに登録されています"
)
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.profiles_repo.create_profile_for_user(
profile_create=ProfileCreate(user_id=created_user["id"])
)
return await self.populate_user(user=UserInDB(**created_user))
# ここまで
async def authenticate_user(
self, *, email: EmailStr, password: str
) -> Optional[UserInDB]:
user = await self.get_user_by_email(email=email, populate=False) # 更新
if not user:
return None
if not self.auth_service.verify_password(
password=password, salt=user.salt, hashed_pw=user.password
):
return None
return user
# 追加ここから
async def populate_user(self, *, user: UserInDB) -> UserInDB:
return UserPublic(
**user.dict(),
profile=await self.profiles_repo.get_profile_by_user_id(user_id=user.id)
)
# ここまで
重要となるのは新しく実装した populate_user メソッドで、profiles_repo を利用してユーザーのプロファイルを UserPublic モデルにアタッチします。get_user_by_email と get_user_by_username の両方のメソッドに単にUserInDBモデルを返すのか、それともpopulate_userメソッドを呼び出した結果を返すのかを決定する新しいpopulateパラメータを追加しました。
authenticate_user メソッドのようにユーザのプロファイルが不要な場合や、実際にユーザのパスワードやソルトにアクセスしたい場合には populate=False を設定することでUserInDBモデルのみを返すことができます。
この変更により適切にアプリは動作しますが以前作成したテストが失敗してしまいます。
主な原因は認証サービスにあるので services/authentication.py を開いて次のように更新します:
...省略
class AuthService:
...省略
def create_access_token_for_user(
self,
*,
user: Type[UserBase], # 更新
secret_key: str = str(SECRET_KEY),
audience: str = JWT_AUDIENCE,
expires_in: int = ACCESS_TOKEN_EXPIRE_MINUTES,
) -> str:
if not user or not isinstance(user, UserBase): ＃更新
return None
jwt_meta = JWTMeta(
aud=audience,
iat=datetime.timestamp(datetime.utcnow()),
exp=datetime.timestamp(datetime.utcnow() + timedelta(minutes=expires_in)),
)
jwt_creds = JWTCreds(sub=user.email, username=user.username)
token_payload = JWTPayload(
**jwt_meta.dict(),
**jwt_creds.dict(),
)
print(jwt.encode(token_payload.dict(), secret_key, algorithm=JWT_ALGORITHM))
access_token = jwt.encode(
token_payload.dict(), secret_key, algorithm=JWT_ALGORITHM
)
return access_token
...省略アクセストークンを作成する際にこれまでは渡されたユーザーがUserInDBのインスタンスであることを確認していました。
しかし先ほどの変更より今後は必ずしもそうとは限らないのでUserInDBとUserPublicの両方が継承する親クラスであるUserBaseに切り替えています。
これにより、アクセストークンが両方のモデルのインスタンスに対して失敗なく作成されるようになります。
テストにも新しい変更を反映させる必要があるので tests/test_users.pyファイルを開き次のように修正します:
...省略
class TestUserRegistration:
async def test_users_can_register_successfully(
self,
app: FastAPI,
client: AsyncClient,
db: Database,
) -> None:
user_repo = UsersRepository(db)
new_user = {
"email": "foo@bar.com",
"username": "foobar",
"password": "bazquxquux",
}
user_in_db = await user_repo.get_user_by_email(email=new_user["email"])
assert user_in_db is None
# 更新ここから
res = await client.post(
app.url_path_for("users:register-new-user"), json={"new_user": new_user}
)
# ここまで
assert res.status_code == HTTP_201_CREATED
...省略
...省略
async def test_users_saved_password_is_hashed_and_has_salt(
self,
app: FastAPI,
client: AsyncClient,
db: Database,
) -> None:
user_repo = UsersRepository(db)
new_user = {
"email": "nmomo@mail.com",
"username": "nmomo",
"password": "nmomoishedgehog",
}
res = await client.post(
app.url_path_for("users:register-new-user"), json={"new_user": new_user}
)
assert res.status_code == HTTP_201_CREATED
# 更新ここから
user_in_db = await user_repo.get_user_by_email(
email=new_user["email"], populate=False
)
# ここまで
assert user_in_db is not None
assert user_in_db.salt is not None and user_in_db.salt != "123"
assert user_in_db.password != new_user["password"]
assert auth_service.verify_password(
password=new_user["password"],
salt=user_in_db.salt,
hashed_pw=user_in_db.password,
)
...省略get_user_by_emai lメソッドを更新して新しいpopulateパラメータを使用し、両方のテストでそれをFalseに設定して以前期待していた同じ動作を維持しています。
またここまでの変更を反映したUserPublicモデルの新しいprofile属性を除外しています。
最後に新しく登録されたユーザーを返すときに発生する別の問題を解決しにいきます。
api/routes/users.pyファイルを開き変更を加えます:
...省略
@router.post(
"/",
response_model=UserPublic,
name="users:register-new-user",
status_code=status.HTTP_201_CREATED,
)
async def register_new_user(
new_user: UserCreate = Body(..., embed=True),
user_repo: UsersRepository = Depends(get_repository(UsersRepository)),
) -> UserPublic:
created_user = await user_repo.register_new_user(new_user=new_user)
access_token = AccessToken(
access_token=auth_service.create_access_token_for_user(user=created_user),
token_type="bearer",
)
return created_user.copy(update={"access_token": access_token})
...省略登録時にUserPublicモデルを返すようになったので、access_token属性を新しいトークンで更新するだけで、そのユーザを返すことができます。
テストを再度実行すると、すべて合格しているはずです！
これでAPIから返されたユーザーにプロフィールがアタッチされるようになりました。
http://localhost:8000/docs で試してみてください。
プロフィールの更新
次にユーザーが自分のプロファイルを更新できることを確認しましょう。
test_profiles.pyに新しいテストクラスを作成します:
...省略
class TestProfileManagement:
@pytest.mark.parametrize(
"attr, value",
(
("full_name", "Nmomos Hedgehog"),
("phone_number", "111-222-3333"),
("bio", "This is a test bio"),
(
"image",
"https://nmomos.com/wp-content/uploads/2019/07/cropped-img_0728.jpg",
),
),
)
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),
("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
これでテストを実行するとNoneを返していることで安定のTypeErrorが発生するので api/routes/profiles.pyファイルを開き、PUTルートを以下のように更新します:
...省略
@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),
profiles_repo: ProfilesRepository = Depends(get_repository(ProfilesRepository)),
) -> ProfilePublic:
updated_profile = await profiles_repo.update_profile(
profile_update=profile_update, requesting_user=current_user
)
return updated_profile現在は存在しない update_profile メソッドを呼び出して必要な更新情報を更新されるユーザーとともに渡しているだけです。
リポジトリ側にこれを実装します:
...省略
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):
...省略
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)
お疲れ様でした、これでテストが全てパスします！
まとめ
APIがようやく形になってきてMVPの最後の仕上げをする準備ができました。
次の投稿ではユーザーがお気に入りのハリネズミに予約を申し込むため関連するルートとリポジトリメソッドをリファクタリングします。
作成したコードはGitHubにアップしています。
part9ブランチが対応しています。
https://github.com/Nmomos/hedgehog-reservation/tree/part9