3
\$\begingroup\$

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!

\$\endgroup\$

1 Answer 1

Reset to default
5
\$\begingroup\$

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.

\$\endgroup\$

You must log in to answer this question.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.