CNLearn FastAPI 4 - CRUD User
So today we’ll have a look at the next part of our project, the base class for our CRUD operations. We will be basing it off this with a few additional concepts from here.
So what is the repository pattern? It’s just an abstraction to make us think we have all our database objects in memory. We don’t have to know how the objects are created, read, updated or deleted; we let the repository pattern deal with that.
So let’s discuss this. What happens here? Let’s go one by one. Once we discuss their implementation, we’ll look at our own.
Let’s start by discussing the from app.db.base_class import Base import. In the base_class file:
from typing import Any
from sqlalchemy.ext.declarative import as_declarative, declared_attr
@as_declarative()
class Base:
id: Any
__name__: str
# Generate __tablename__ automatically
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()
SQLAlchemy has two ways of mapping styles: imperative and declarative. What does mapping mean though? Mapping is essentially associating user-defined Python classes with their database tables, and objects of those classes with rows in those tables. In the declarative style, we usually construct a base class using the declarative_base() function. Normally, you’d do something like Base = declarative_base() and then have other classes inherit from that Base class. In the code above, the first create a Base class and have the @as_declarative decorator. What does the decorator do? It does the same thing but at the same time our Base class has a few more attributes. What are they? The first one is the id. Any class that inherits from the Base declarative class will have this id attribute. At the same time, there’s also a tablename method that is decorated using the declared_attr. Why is it decorated? Well, that decorator turns this class method into a property that contains the __tablename__. Suppose we look at our User model.
class User(Base):
id = Column(Integer, primary_key=True, index=True)
full_name = Column(String, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean(), default=True)
is_superuser = Column(Boolean(), default=False)
That means the User tablename will be “user”, which, as we determined in the previous post, maybe isn’t such a great idea. So, for the User model, we will actually overwrite the default tablename attribute (for all the other ones it’s fine.) So our User model will be:
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
full_name = Column(String, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean(), default=True)
is_superuser = Column(Boolean(), default=False)
Also, as I am writing this I am realising that we didn’t actually look at the User model before but we did create the table. oops. And in fact, at the end of this post, we’ll cover a better way of using Alembic with SQLAlchemy. Why’s that? Well, in the previous post regarding Alembic, we manually created the migrations. That is not fun and seems like double the work. We’ll actually downgrade our database, and then auto generate migrations :) yay
But before we do that, let’s look at the CRUDbase code here. What is CRUD you ask? They’re fundamental database storage functions: Create, Read, Update and Delete. Let’s start by looking at the imports. Keep in mind that this project will be fully types (and in a future post we’ll look at having a pre-commit hook but also some CI on GitHub using GitHub actions).
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.db.base import Base
ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
The first few lines are all related to Typing. We import TypeVar which is a type variable. What does it mean? Well, in ModelType = TypeVar("ModelType", bound=Base) we create a ModelType type variable (that we can use for type hints). If it was only ModelType = TypeVar("ModelType") and nothing else, that type variable could have been anything (i.e. same as Any). If we had ModelType = TypeVar("ModelType", str, int), the ModelType could have only been a string or an integer. But that’s not what we have here. We have a bound=Base parameter. What does that do? To quote the documentation,
By default type variables are invariant. Alternatively, a type variable may specify an upper bound using bound=. This means that an actual type substituted (explicitly or implicitly) for the type variable must be a subclass of the boundary type
In other words, they have to be subtypes of whatever the upper bound is, in this case the SQLAlchemy declarative base class Base. This will indicate that our ModelType will essentially base an SQLAlchemy mapped class type. Then, we also have the Create and Update Schema Types which are also upper bound to BaseModel, where BaseModel is the pydantic BaseModel class. Keep in mind that throughout the project we will be referring to SQLAlchemy models as models and to pydantic models as schemas.
We then have our CRUDBase class definition. Let’s go line by line. We start with:
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
Inheriting from Generic? Well, this is how we indicate that our CRUDBase is a generic type and the type variables it takes in are ModelType, CreateSchemaType and UpdateSchemaType. This allows us to type hint any class that inherits from this in a very generic way. Let’s quickly jump to the CRUDUser class in (app.crud.crud_user) where we can see that it’s defined as class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]), i.e. CRUDUser inherits from CRUDBase AND it’s saying that the ModelType is User, CreateSchemaType is UserCreate and and UpdateSchemaType. We’ll get to that shortly so let’s keep looking at CRUDBase, starting with the ``init` function.
def __init__(self, model: Type[ModelType]):
self.model = model
So when the class gets instantiated (or any classes inheriting from this one), the model instance attribute gets set to the model class used (whether it’s User or another one). Why is the type hint Type[ModelType]? Let’s have a look at how the CRUDUser class )which we will get to…eventually) is instantiated.
user = CRUDUser(User)
That means that the user.model attribute is set to the User model class. If the type hint for model was ModelType, then it would expect an instance of that class. Since we want to refer to the class itself, we use Type[ModelType]. Init sorted, let’s have a look at the CRUD methods (create, read, update, delete) as well as get_multi method the CRUDBase provides. We will begin with the get (i.e. Read) method:
async def get(self, db: AsyncSession, id: Any) -> Optional[ModelType]:
results = await db.execute(select(self.model).where(self.model.id == id))
return results.scalars().first()
We use the Session object to communicate with the database by invoking a Select with the ORM. The execute command executes the SQL expression construct and returns a Result object. We then return the first result or None if there aren’t any. It’s worth noting that because we are using the id, which is the primary key, we are guaranteed that there will either be one result or None, there won’t be multiple ones.
The get_multi method has a similar construct with the execute command and the select command, except we also have offset and limit arguments. Nothing special if you’ve used SQLAlchemy in the past and even if you haven’t, the notation is relatively straightfoward.
async def get_multi(
self, db: AsyncSession, *, skip: int = 0, limit: int = 100
) -> List[ModelType]:
results = await db.execute(select(self.model).offset(skip).limit(limit))
return results.scalars().all()
What is more interesting, however, is the function signature self, db: Session, *, skip: int = 0, limit: int = 100. The notatation was introduced in PEP 3102 and says that any arguments after the empty * are keyword-only arguments. The session object can be passed to the method as a positional argument, but skip and limit can only be passed as keyword arguments, e.g. .get_multi(db, skip=skip, limit=limit). Please note that since they do have default values, we do not need ot send those in. Finally, we use the .all() method on the Result object which returns a List of Row
The create method takes in a similar session but it also has a keyword only argument. This time, however, it doesn’t have a default value. Why? Because that CreateSchemaType might be a UserCreateSchemaType, a SOMETHINGELSECreateSchemaType, etc. Keep in mind that they can be subtypes of the CreateSchemaType which is itself upper bound to BaseModel -> i.e. they are all Pydantic schemas (I’m not saying Model so we don’t confused them with the SQLAlchemy models).
async def create(self, db: AsyncSession, *, obj_in: CreateSchemaType) -> ModelType:
obj_in_data = jsonable_encoder(obj_in)
db_obj = self.model(**obj_in_data) # type: ignore
db.add(db_obj)
await db.commit()
await db.refresh(db_obj)
return db_obj
We pass that schema object through FastAPI’s jsonable_encoder in order to convert all the data types to JSONable ones before saving in the database. We use the .add methodf of the db (session object) in order to place instances of our models in the session. This is equivalent to the INSERT sql query. Please note that the instance is held in the session until the next flush, which occurs when the commit() method is called. Why do we then call the refresh() method? We do so in order to get the latest attributes on the instance (in case there were some that were changed/created in the DB itself). Please note that since we are delaing with async SQLAlchemy, all these have an await before them.
We can also update instances in the database. Similar behaviour, except we can call the function with the SQLAlchemy model instance and then with either a dict of data or a Pydantic schema instance. We first create a JSON dict of the current data in the model instance. Then, if the obj_in we are passing is a dict, we leave it as it is. Otherwise, we call the dict model on the Pydantic schema instance, with exclude_unset=True which causes fields that were not set to be excluded from the dictionary that gets generated/created.
async def update(
self,
db: AsyncSession,
*,
db_obj: ModelType,
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
) -> ModelType:
obj_data = jsonable_encoder(db_obj)
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.dict(exclude_unset=True)
for field in obj_data:
if field in update_data:
setattr(db_obj, field, update_data[field])
db.add(db_obj)
await db.commit()
await db.refresh(db_obj)
return db_obj
We then iterate over the fields in the (current) db object dict. If they are also in the update_data dictionary (i.e. the things we want to change), we change them on the database object. We add the object to the session, commit, refresh and return it.
Finally we also have a remove/delete method. We get the object from the database (its existance is verified before this method is called), we delete, commit, and return what was deleted.
async def remove(self, db: AsyncSession, *, id: int) -> ModelType:
obj = await db.get(self.model, id)
await db.delete(obj)
await db.commit()
return obj
User Schema and User CRUD
Ok we have covered this CRUDBase but it’s all generic. When/where do we actually use it? Well, let’s look at the app.crud.crud_user file at its usage with the User model and schema. I guess that in order to do that, we should first see the User (Pydantic) schema in app.schemas.user. Remember that our User SQLAlchemy model was:
User SQLAlchemy Model
id = Column(Integer, primary_key=True, index=True)
full_name = Column(String, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean(), default=True)
is_superuser = Column(Boolean(), default=False)
User Pydantic Schemas
Our UserBase Pydantic schema has:
class UserBase(BaseModel):
email: Optional[EmailStr] = None
is_active: Optional[bool] = True
is_superuser: bool = False
full_name: Optional[str] = None
It might be strange seeing a UserBase with only four properties, 3 of which are Optional, 2 of which have a default value of None. Surely some can’t be optional and/or None? This is where inheriting the UserBase schema comes in. For example, the Pydantic Schema that will be used when “creating” an user will be UserCreate:
class UserCreate(UserBase):
email: EmailStr
password: str
Note that for this one, email now becomes mandatory and we also add a password field (also mandatory). That makes sense as we’d need both email and password to create an account. We could pass in a full_name as well, if we wanted to. is_superuser is by default False (since we don’t want to expose that to allow creation of more superuser accounts). Similarly, is_active is True by default as we’d want the user account to be active when it’s created (if we want to have some kind of activation, we might change that to False later on).
Similarly, the UserUpdate schema is used:
class UserUpdate(UserBase):
password: Optional[str] = None
when we are updating user data. We might not always want to change password and that is why it is optional.
Finally, there is one more schema that a further one will derive from. The UserInDBBase schema:
class UserInDBBase(UserBase):
id: Optional[int] = None
class Config:
orm_mode = True
What is this orm_mode and what is that Optional id? Well, Pydantic schemas can be used with ORM objects (in this case the SQLAlchemy models we use). You can find more information here. Those have have an id.
Finally, the schema inheriting from UserInDBBase:
# Additional properties to return via API
class User(UserInDBBase):
pass
It will be used in the FastAPI response model as it will have an id, and all the UserBase fields.
User Crud
Let’s look at the CRUDUser class.
from typing import Any, Dict, Optional, Union
from sqlalchemy.orm import Session
from sqlalchemy import select
from app.core.security import get_password_hash, verify_password
from app.crud.base import CRUDBase
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
...
user = CRUDUser(User)
First of all, it inherits from CRUDBase which we looked at earlier. Then, it inherits all the methods of CRUDBase but at the same time it is overwriting some. First of all, we might not want to get a user by their ID. It’s not like we’ll ask the User: hey Bobby, what’s your database ID? How should we find them? By email!
async def get_by_email(self, db: AsyncSession, *, email: str) -> Optional[User]:
result = await db.execute(
select(User).where(User.email == email)
)
return result.scalars().first()
What about creating a user?
async def create(self, db: AsyncSession, *, obj_in: UserCreate) -> User:
db_obj = User(
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
full_name=obj_in.full_name,
is_superuser=obj_in.is_superuser,
)
db.add(db_obj)
await db.commit()
await db.refresh(db_obj)
return db_obj
Note that we are not saving the password, we are saving the hashed password. We add the object, refresh (getting the ID) and return the object. How do we update the info for a User?
async def update(
self, db: AsyncSession, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
) -> User:
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.dict(exclude_unset=True)
print(update_data)
if update_data.get("password"):
hashed_password = get_password_hash(update_data["password"])
del update_data["password"]
update_data["hashed_password"] = hashed_password
print(obj_in.__dict__)
return await super().update(db, db_obj=db_obj, obj_in=update_data)
This method can actually either take a UserUpdate (Pydantic) schema or a dictionary of keys and values. If it’s a dictionary instance, it remains like that. Otherwise, we create a dictionary from the schema (excluding any keys where the value is None so we don’t accidentally remove info). Remember that the UserUpdate password was optional? Well, if it is set, we change the password by creating a password hash of the new password, removing the password from the update dict and sending it to the CrudBase class.
How do we log in users -> i.e. how do we authenticate them?
async def authenticate(self, db: AsyncSession, *, email: str, password: str) -> Optional[User]:
user = await self.get_by_email(db, email=email)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
We check if there’s an user with that email. We then use a helper function to calculate the hash of the password the user entered at login and compare with the hashed password in the database. If they match, the user is authenticated.
Finally, there are two useful methods:
async def is_active(self, user: User) -> bool:
return user.is_active
async def is_superuser(self, user: User) -> bool:
return user.is_superuser
The first one simply returns if the user is active and the second one whether is a super user.
Authentication and User Related Endpoints
So far, we’ve looked at the repository pattern, its implementation for the User model, but what about using all of these? It’s time to look at how registration and authentication will work.
Registration Endpoints
Finally we’re back to some FastAPI stuff. We will look at the registration endpoint, but it will not be publicly accesible. In fact, it won’t be used for quite a while until I’m ready to allow users on this. Still, what do yu think it should be able to do? It needs to connect to the database, check if the email is not already in use, and call our UserCRUD create method.
In our app.api.v1.router.registration.py file, let’s import a few useful things: APIRouter, Body, Depends, HTTPException. Then let’s create the endpoint for registration:
import logging
from typing import Any
from fastapi import APIRouter, Body, Depends, HTTPException
from pydantic import EmailStr
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.dependencies import database
from app import schemas
from app.crud import user
router: APIRouter = APIRouter()
logger = logging.getLogger(__name__)
@router.post("/register", response_model=schemas.User)
async def create_user(
*,
db: AsyncSession = Depends(database.get_async_session),
password: str = Body(...),
email: EmailStr = Body(...),
full_name: str = Body(None),
) -> HttpResponse:
"""
Endpoint for registering a new user
"""
# first, check if such a user does not already exist
existing_user = await user.get_by_email(db, email=email)
if existing_user:
raise HTTPException(
status_code=400,
detail="This email is already in use."
)
user_in = schemas.UserCreate(password=password, email=email, full_name=full_name)
new_user = await user.create(db, obj_in=user_in)
return new_user
Let’s go bit by bit. We create an APIRouter (that we will import higher up in the hierarchy), and then we create a “register” path for it. We’ll come back to response_model shortly and instead let’s look at what our create_user function takes as input. First of all, it only accepts keyword arguments and it accepts db, password, email and full_name:
- db keywork argument is actually a Dependency injection which specifies that the dependency is a certain database.get_async_session (we’ll look at this soon)
- password, email and full_name are manually specified as Body. Why are we specifying this manually instead of using something like UserCreate? Well, we don’t want the API to accept anything except those fields. We certainly wouldn’t want them to be able to set is_active at registration and definitely not is_superuser…
What is this Dependency Injection we talk of? It’s basically us telling the function that in order for it to work, we need that dependency function to run and that we will use the result. In this case, our db really refers to a SQLAlchemy async session.
The actual function looks like:
async def get_async_session(request: Request) -> AsyncSession:
async_session_maker = request.app.state._db
async with async_session_maker() as async_session:
yield async_session
It receives the request, accesses the app from it (all requests have access to the overall App instance as an attribute). Remember how we had a startup event handler
app.add_event_handler("startup", database_start_app_handler(app))
in which we saved the asyncsession creator to app.state._db = async_session?
Well in the dependency specified above, we actually yield a session we can use. We use yield because that way the session will be closed at the end.
Going back to our create user endpoint, we must first check if there isn’t such a user already. To do what, we use the UserCRUD get_by_email method. If a user is found, that means the email is already in use so we return a 400 Bad Request HTTP Response (as an HTTPException). If, however, there is no existing user, we use the UserCreate schema, passing in password, email and full_name to create a user_in instance. Keep in mind that in this UserCreate schema instance, is_active will automatically be set to True, and is_superuser will be set to None. We then call the user.create(db, obj_in=user_in) method, which then createa a User model instance, passing in all the values that we have. Note we don’t have to pass in id (since that will be returned by the DB) and we also don’t have to pass in is_active, because by default it will have a value of True. The other thing to note is that we save the hashed_password, not the actual password (NEVER EVER EVER DO THAT). In order to get the hash of the password, we have a get_password_hash function defined in app.core.security. We won’t go into detail about that now, I’ll dedicate the next post on passwords and security and encryption and whatever, but it’s sufficent to say that it encrypts the password. The new model instance gets added to the session and will get written to the database next time we flush, and we do that by committing.
In
new_user = await user.create(db, obj_in=user_in)
return new_user
we are getting the new user and then returning that. But what happens then? We’re simply returning a SQLAlchemy model instance. How does FastAPI know how to turn that into the JSON response we get? Remember that our schemas.User inherits from UserInDBBase which has orm_mode set to True. This tells Pydantic to read the data even if it’s an ORM model (i.e. some kind of object with attributes).
And you know what? This is where we’ll stop adding new things to the app here…well, yes and no because..it’s time for tests :)
How do we test our FastAPI app? Since we’re doing everything async, it’s a little trickier but doable. This post uses some very useful information from here but also from a few various StackOverflow answers.
Setting up Testing
Creating a testing database and creating our fixtures
First thing we will do is create a testing database.
CREATE DATABASE cnlearn_testing
Some people prefer to delete the database entirely and create it again, that introduces some unnecessary complexity in my opinion. Maybe we’ll go down that route one day, but for now, we will simply destroy and recreate everything in the test database.
We will then create a tests directory in the root directory and add a conftest.py file in there. This will store our fixtures that will be used in the entire testing suite. What do we need?
Well, we need to be able to destroy and recreate everything in the database. What do you mean? We will create the tables, run the migrations (upgrade) at the beginning of our testing session and reverse the migrations (downgrade) at the end of our testing session. We will have a fixture called apply_migrations in tests/conftest.py.
@pytest.fixture(scope="session")
def apply_migrations():
os.environ["TESTING"] = "1"
config = Config("alembic.ini")
alembic.command.upgrade(config, "head")
yield
alembic.command.downgrade(config, "base")
We’ll discuss the TESTING environment variable in just a second. The more important part is reading our Alembic configuration file (alembic.ini), we then run the upgrade command (to run our migrations “forward”) and yield. What does yield do? It means that for example, let’s say we use this fixture in a test. Everything until the yield point runs, the test runs, goes back to the fixture, everyting after the yield command runs. This way, our downgrade command (to “reverse” our migrations) only runs after the test finishes. In this case, however, the “test” refers to the entire test suite since this fixture is run once a session.
Why are we setting TESTING as 1 in our environment? Well, because depending on whether it’s 0 or 1 our app will use the main or the testing database respectively. You might be asking: where does this happen? Remember our open_postgres_database_connection function that gets called when the FastAPI app starts? We added this in app/db/connections.py:
async def open_postgres_database_connection(app: FastAPI) -> None:
# these are configured in the settings module
ASYNC_URI: str = settings.SQLALCHEMY_POSTGRES_URI.replace(
"postgresql", "postgresql+asyncpg", 1)
if settings.TESTING:
ASYNC_URI += "_testing"
...
and obviously a TESTING field to our settings in app/settings/base.py.
# PostgreSQL database settings
POSTGRES_USER: str
POSTGRES_PASSWORD: str
POSTGRES_SERVER: str
POSTGRES_DB: str
POSTGRES_PORT: int
TESTING: int = 0 # if this is 1, we will connect to the testing database
SQLALCHEMY_POSTGRES_URI: Optional[PostgresDsn] = None
Ok we have created our migrations fixture. Where is it used? Well, it’s used in our app fixture.
@pytest.fixture(scope="session")
def app(apply_migrations: None) -> FastAPI:
from app.server import create_application
return create_application()
Do you see how we are using the apply_migrations fixture as a dependency in our app fixture? Although we are not using it directly, it runs as it’s part of the fixture for it. Finally, let’s create an async client test app, which, maybe you guessed, uses the app fixture (which itself uses the apply_migrations fixture). The next part is taken from here.
@pytest_asyncio.fixture
async def client(app: FastAPI) -> AsyncClient:
async with LifespanManager(app):
async with AsyncClient(
app=app,
base_url="http://testserver",
headers={"Content-Type": "application/json"}
) as client:
yield client
Ok there are a few things happening in there. We also need a few more packages:
pip install asgi_lifespan
pip install httpx
pip install pytest_asyncio
Back to the test. What does the fixture do? It creates an AsyncClient wrapped around our app that we can use.
Finally, we are going to create two more root fixtures: one that will provide us with a SQLAlchemy Async Session and one that will delete all users from a table. We will get to why shortly. Let’s look at the first one.
@pytest_asyncio.fixture
async def get_async_session(client: AsyncClient, app: FastAPI) -> AsyncSession:
async_session_maker = app.state._db
async with async_session_maker() as async_session:
yield async_session
Why do we need this? Aren’t we just going to use the client fixture for our tests? Yes, for the endpoints. I do, however, want to test everything including our crud base and user crud we created. In order to do that, I will need an async_session. Why do we need to request client in there as well? I won’t go into the details too much (that will be in my series on Async which I will one day get back to :( I start too many things, I know)) but if you call the async with separately, it will default to being in a new async loop instead of the one client is in…annoying, I know.
Lastly, we also have a fixture that will clean all users in the table.
@pytest_asyncio.fixture
async def clean_users_table(get_async_session: AsyncSession):
yield
db: AsyncSession = get_async_session
users_deleted: CursorResult = await db.execute(delete(User))
await db.commit()
Why do we need that? Well, I like having clean tests at the beginning of a test, it will reduce test flakiness considerably. If we have this fixture in our test, it will yield directly to the test, run the test, and then run the rest of the code that will delete all users from the table. We will probably create more of these cleanup functions as we add more tests and models.
Have I bored you enough with the fixtures? I have? Ok let’s look at one more. Before we do that, however, I just wanted to say that I like to split my test folder structure to match what our app directory structure is like. Since we are going to test crud_user, we will have a test file path called: tests/crud/test_crud_user.py. In the same folder, we will have a conftest.py file (path tests/conftest.py) with the following factory as a fixture.
@pytest.fixture
def user_schema() -> Callable[..., schemas.UserCreate]:
"""
This fixture returns a User schema object.
"""
def _make_user_schema(*, password: str, email: str, full_name: str):
return schemas.UserCreate(
password=password,
email=email,
full_name=full_name
)
return _make_user_schema
I think you are ready for our first tests: of the crud_user (which will also test our base crud class).
Testing our CRUDUser
We start by importing everything we need:
from typing import Callable, Optional
from unittest import mock
from sqlalchemy.ext.asyncio import AsyncSession
import pytest
from app.crud import user
from app.models.user import User
from app import schemas
and let’s look at the function signature first. It will be a function that tests all the methods at once, since I’m a bit lazy to write individual tests for all of them. In fact, I wrote the test to more or less resemble a typical usage. We might first check that there are no users in there. Then, we will create an user (and check the password got hashed). We will get the user to make sure it got created. We will then request all the users to see how many there are. We will then update our user’s email. We will then update our user’s password (to check the hashing does work on updating). We’re going to pretend we forgot their id, and try to find them by email. We will then unsuccessfully “authenticate” twice (bad email, bad password) before authenticating successfully. Finally, the user decided they don’t like the website so requested their user to be deleted. We will comply. That will be what happens in the test. Let’s look at the signature, however, as that’s more interesting.
@mock.patch("app.crud.crud_user.verify_password")
@mock.patch("app.crud.crud_user.get_password_hash")
@pytest.mark.asyncio
async def test_crud_user(
mocked_get_password_hash: mock.MagicMock,
mocked_verify_password: mock.MagicMock,
get_async_session: AsyncSession,
user_schema: Callable[..., schemas.UserCreate],
):
We have a test_crud_user async test function. Since it’s async, we need to decorate it with the @pytest.mark.asyncio flag. We also want to keep our tests as isolated as possible: in CRUDUser, the from app.core.security import get_password_hash, verify_password functions get used. Let’s mock them so we can verify the password does get hashed when saving, when updating, and that verify_password does check the hash of the password on authentication. Notice first I patch get_password_hash (1st mock) and then on top, I patch verify_password (2nd mock). This is the order they will FIRST appear in the arguments list of the test function. The innermost mock appears first, and outer ones appear next. The first two arguments are, therefore, patch mocks. The third argument is a fixture that will be taken from the root tests/conftest.py file. The fourth one comes from the tests/crud/conftest.py file. The get_async_session will provide us with a SQLAlchemy AsyncSession and the user_schema with a UserCreate schema we can send to the database.
@mock.patch("app.crud.crud_user.verify_password")
@mock.patch("app.crud.crud_user.get_password_hash")
@pytest.mark.asyncio
async def test_crud_user(
mocked_get_password_hash: mock.MagicMock,
mocked_verify_password: mock.MagicMock,
get_async_session: AsyncSession,
user_schema: Callable[..., schemas.UserCreate],
):
# let's first call user get with nothing in there
no_user: Optional[User] = await user.get(get_async_session, id=1)
assert no_user is None
mocked_get_password_hash.return_value = "MockedPassword"
user_in: schemas.UserCreate = user_schema(password="amazing", email="user@email.com", full_name="Fake Name")
new_user: User = await user.create(get_async_session, obj_in=user_in)
assert new_user.email == "user@email.com"
assert new_user.full_name == "Fake Name"
assert new_user.hashed_password == "MockedPassword"
assert new_user.is_active == True
assert new_user.is_superuser == False
# let's get the new_user.id since that can vary depending on what other tests ran first
user_id: int = new_user.id
# now let's call user get again and we should have one with the same id
existing_user: Optional[User] = await user.get(get_async_session, id=user_id)
assert isinstance(existing_user, User)
# let's call get_multi and see that there are in fact 1 user(s)
users: list[User] = await user.get_multi(get_async_session)
assert len(users) == 1
# you want to change your email? ok let's do that
user_schema_in = schemas.UserUpdate(email="newuser@email.com")
updated_user: User = await user.update(get_async_session, db_obj=new_user, obj_in=user_schema_in)
assert updated_user.email == "newuser@email.com"
# you even want to change your password? ok let's generate a new mock return_value
mocked_get_password_hash.return_value = "MockedPassword2"
user_schema_in = schemas.UserUpdate(password="getsoverwrittenbythehash")
updated_user: User = await user.update(get_async_session, db_obj=updated_user, obj_in=user_schema_in)
assert updated_user.hashed_password == "MockedPassword2"
# do you think we can find you by email? let's see
user_by_email: Optional[User] = await user.get_by_email(get_async_session, email="newuser@email.com")
assert isinstance(user_by_email, User)
assert user_by_email.email == "newuser@email.com"
assert user_by_email.id == user_id
user_no_such_email: Optional[User] = await user.get_by_email(get_async_session, email="fake@email.com")
assert user_no_such_email is None
# let's check whether the user is active and is a superuser. by default, it will be True and False respectively
assert await user.is_active(updated_user) is True
assert await user.is_superuser(updated_user) is False
# let's log our user in. we will enter an email that does not exist first
no_logged_in_user: Optional[User] = await user.authenticate(get_async_session, email="fake@email.com", password="asd")
assert no_logged_in_user is None
# ok now we forgot our password momentarily
mocked_verify_password.return_value = False
incorrect_password_no_user: Optional[User] = await user.authenticate(get_async_session, email="newuser@email.com", password="asd")
assert incorrect_password_no_user is None
assert mocked_verify_password.called
# ooooh I remember the password now
mocked_verify_password.return_value = True
correct_password_user: Optional[User] = await user.authenticate(get_async_session, email="newuser@email.com", password="doesntmatter")
assert isinstance(correct_password_user, User)
assert mocked_verify_password.called
# you want to leave us? :( well, I am sorry to hear that
removed_user: Optional[User] = await user.remove(get_async_session, id=user_id)
assert isinstance(removed_user, User)
# let's check that there's no one left
assert len(await user.get_multi(get_async_session)) == 0
The rest of the stuff is hopefully pretty self explanatory.
Testing our Registration Endpoint
Now, we write an actual FastAPI test rather than of our underlying code.
from typing import Callable, Any
from fastapi import FastAPI
from httpx import AsyncClient, Response
import pytest
@pytest.mark.asyncio
async def test_create_user(client: AsyncClient, app: FastAPI, clean_users_table: Callable[..., None]):
create_user_url: str = app.url_path_for("user:create-user")
response: Response = await client.post(
url=create_user_url,
json={
"password": "interesting",
"email": "unique@email.com",
"full_name": "Uniquely Interesting",
}
)
json_response: dict[str, Any] = response.json()
assert json_response["email"] == "unique@email.com"
assert json_response["is_active"] is True
assert json_response["is_superuser"] is False
assert json_response["full_name"] == "Uniquely Interesting"
assert isinstance(json_response["id"], int)
We need the client, app and clean_users_table fixtures from our root tests/conftest.py file. Why? Well, the client is for interacting (get, post, whatever) with our app. The app is because I will get the url of the path we will be posting to that way. The clean_users_table in order to delete the user we are creating in our test. An interesting thing I did was with this:
create_user_url: str = app.url_path_for("user:create-user")
Where does the “user:create-user” come from? Well, I actually named my endpoint path in app/api/v1/routes/registration.py. By default, the name is the name of the function (create_user).
@router.post("/register", response_model=schemas.User, name="user:create-user")
I am doing Django all day every day for my job (well not just, but a big part) and so I like using that namespaced convention.
You might also wonder, why did I not just do:
assert response.json() == {
"email": "unique@email.com",
"is_active": True,
"is_superuser": False,
"full_name": "Uniquely Interesting",
"id": 1,
}
Well, because of the ID. If this test runs first, ID will be 1. If it runs at some point, even if users are deleted, the ID will not be reused and it might be 3, or 99. Who knows…aaaaaand I think that’s it. I only started this post forever go, but now I’ve had a bit more time beside the job so I hope to post more. I really want to document the whole process :( but that does slow down the time it takes me to complete it. But I’ll be better, I promise. In the next post we will create some stuff related to our testing, pre-commit and all of these things on GitHub with GitHub Actions (since we have actual tests now with database now).
The commit for this post is here.