Users CRUD in fastapi using pymongo

Question

I am building users API with CRUD operation in fastapi and i'd love to hear feedback about it. I have been exploring fastapi in the past weeks and im trying to create API with best practice (scalable - no pagination yet, robust, effective and efficient)

route.py

from fastapi import APIRouter, Depends
from users.serializer import User
from users.schema import UserUpdate, UserResponse, UserCreate, UserFilter
user_router = APIRouter(
 prefix="/users",
 tags= ["Users"]
 )
@user_router.get("/",response_model=list[UserResponse])
def get_users():
 return User.get_all()
@user_router.get("/filter", response_model=list[UserResponse])
def filter_user(query: UserFilter = Depends()):
 filter = User.filter(dict(query))
 return filter
@user_router.get("/{id}", response_model=UserResponse)
def get_user(id: str):
 user = User.get(id)
 return user
@user_router.post("/", response_model=UserResponse)
def save_user(user: UserCreate):
 new_user = User(dict(user))
 return new_user.create()
@user_router.put("/{id}", response_model=UserResponse)
def update_user(id: str, updated_user: UserUpdate):
 user = User(updated_user)
 return user.update(id)
@user_router.delete("/{id}")
def delete_user(id: str):
 user = User.delete(id)
 return user

schema.py

from typing import Optional
from pydantic import BaseModel, EmailStr, model_validator
from datetime import datetime
class UserCreate(BaseModel):
 username: str
 email: EmailStr
 password: str
 first_name: str
 last_name: str
 role: str
 status: str
 created_at: Optional[datetime] = datetime.now()
 updated_at: Optional[datetime] = datetime.now()
 class Config:
 orm_mode: True
class UserUpdate(BaseModel):
 username: str
 email: EmailStr
 first_name: str
 last_name: str
 role: str
 status: str
 created_at: Optional[datetime] = datetime.now()
 updated_at: Optional[datetime] = datetime.now()
 class Config:
 orm_mode: True
class UserResponse(BaseModel):
 id: str
 username: str
 email: EmailStr
 first_name: str
 last_name: str
 role: str
 status: str
 created_at: Optional[datetime] = datetime.now()
 updated_at: Optional[datetime] = datetime.now()
class UserLogin(BaseModel):
 username: str
 password: str
class UserFilter(BaseModel):
 username: Optional[str] = None
 email: Optional[EmailStr] = None
 first_name: Optional[str] = None
 last_name: Optional[str] = None
 role: Optional[str] = None
 status: Optional[str] = None

serializer.py

from utils.passwords import PasswordHandler
from gateways.database import DatabaseGateway
from fastapi import HTTPException, status
collection_name = "users"
class User:
 def __init__(self, data) -> None:
 self.data = data
 def create(self):
 password_handler = PasswordHandler()
 self.data["password"] = password_handler.hash(self.data["password"])
 user = DatabaseGateway(collection_name)
 return user_serializer(user.save_document(self.data))
 
 def update(self, id):
 del self.data.created_at
 user = DatabaseGateway(collection_name)
 return user_serializer(user.update_document(id, self.data))
 
 def delete(id):
 user = DatabaseGateway(collection_name)
 result = user.delete_document(id)
 if result:
 return user_serializer(result)
 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found!")
 
 def filter(filter):
 clean_filter = {k: v for k, v in filter.items() if v is not None}
 users = DatabaseGateway(collection_name)
 filtered_document = users.filter_document(clean_filter)
 serialized_users = serialize_list(filtered_document)
 if serialized_users:
 return serialized_users
 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found!")
 def get_all():
 users = DatabaseGateway(collection_name)
 serialized_users = serialize_list(users.get_collection())
 if serialized_users:
 return serialized_users
 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="users not found!")
 
 def get(id):
 user = DatabaseGateway(collection_name)
 result = user.get_document(id)
 if result:
 return user_serializer(result)
 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found!")
 
def user_serializer(data) -> dict:
 user = {
 "id": str(data["_id"]),
 "username": data["username"],
 "email": data["email"],
 "first_name": data["first_name"],
 "last_name": data["last_name"],
 "role": data["role"],
 "status": data["status"],
 "created_at": data["created_at"],
 "updated_at": data["updated_at"]
 }
 return user
def serialize_list(users) -> list:
 return [user_serializer(user) for user in users]

gateway/database.py

from config.database import Database
from bson import ObjectId
from fastapi import HTTPException
class DatabaseGateway:
 def __init__(self, collection_name):
 self.collection_name = collection_name
 def save_document(self, data):
 document = Database.collection(self.collection_name).insert_one(dict(data))
 data["_id"] = str(document.inserted_id)
 return data
 def get_collection(self):
 return Database.collection(self.collection_name).find()
 
 def get_document(self, id:str): 
 vba = ObjectId.is_valid(id)
 if vba:
 return Database.collection(self.collection_name).find_one({"_id": ObjectId(id)})
 
 
 def update_document(self, id, data:dict):
 document = Database.collection(self.collection_name)
 return document.find_one_and_update({"_id": ObjectId(id)}, {"$set": dict(data)}, return_document=True)
 
 def delete_document(self, id:str): 
 vba = ObjectId.is_valid(id)
 if vba:
 return Database.collection(self.collection_name).find_one_and_delete({"_id": ObjectId(id)})
 
 def filter_document(self, filter):
 return Database.collection(self.collection_name).find(filter)

config/database.py

client = MongoClient(config("db_client"))
db = client.fms
# list of modules/collection (users, author, books)
collections = {
 "users": db['users']
}
class Database:
 def collection(collection: str):
 if collection in collections:
 return collections[collection]

I'd love your review or opinion on this, small or big. Thanks a lot!

Answer 1

test suite

There isn't one.

love your review or opinion on this

I don't quite know which way I should approach this. We have a big blob of code, with no def test_foo, def main, or other obvious entrypoint for exercising the target code. I imagine there's a way to run it, but it is not yet obvious to me what way that would be. So, ok, I will just read sequentially, and we'll see what comes of it.

lint

Run this through black, please? The random whitespace variations are just distracting. Thank you.

May as well tack on isort whilst you're at it.

Oh, and thank you kindly for the type annotations, they're lovely.

mutable default

def filter_user(query: UserFilter = Depends()):

I don't know exactly what's going on with Depends(). But it makes me nervous, since a great many python programmers have been tripped up by mutable parameter defaults, for example

def foo(d: dict = {'a': 1, 'b': 2}):

The gotcha is that d is given a default value just once, when we define foo, rather than being given N new dict's for N calls to foo(). This surprises some folks, since updates to that single d value will persist over a great many calls.

tl;dr: Have the signature default to None, and then within the function assign query = query or Depends() to compute a new Depends instance on each call, when needed.

credentials

class UserCreate(BaseModel):
 username: str
 email: EmailStr
 password: str

I don't understand that last line. In the sense that, I don't understand why anyone would store a cleartext password in a database. Data breaches happen. At most, an attacker should only be able to harvest hashes, preferably produced by argon2id.

There is maybe some confusion in this codebase between "password" and "hash"? Also, if a per-user "salt" is being stored, I haven't seen that explicitly called out by the code.

optional

I dislike these:

 created_at: Optional[datetime] = datetime.now()
 updated_at: Optional[datetime] = datetime.now()

Surely it is trivial to impose a NOT NULL constraint on those, right? How hard could it be to ask the DB to fill those in on every INSERT and UPDATE? I have routinely worked with such constraints, across several backend RDBMS vendors.

Similarly for the other tables. It's just too easy to always fill in these timestamps, and they really will prove useful.

nuke created

 def update(self, id):
 del self.data.created_at

I don't understand what's going on there.

This may relate to the whole "non-NULL created constraint" mentioned above.

This record was created at a certain timestamp, that fact is immutable, and I don't get why we'd want to change that.

shadowing

 "id": str(data["_id"]),

It's not obvious to me what's going on here.

Many methods take an id parameter, which shadows the id builtin function, but I bit my tongue on those, it didn't seem troubling.

The prefix _ convention is for a _private helper or variable.

The suffix _ convention is to distinguish our id_ from the builtin id().

Those don't seem very relevant to dict key names, but I am happy to be enlightened.

In user_serializer() I am a little sad that we didn't have enough uniformity to enable just looping over a bunch of key names.

In DatabaseGateway, I confess I have no idea what the TLA vba denotes. We find no helpful # comments.

type stability

In {get,delete}_document, we either return a DB row or None. I find it troubling that we just fall off the end of the function; please replace that with an explicit return None.

Please add docstrings to these methods, or ...) -> None: type annotations, which explain to the caller they will optionally get a DB row.

Additionally, delete_document() really needs a docstring explaining what comes back, since the name suggests a "void" routine one would evaluate just for side effects.