MinimaList: simple shopping list app – intro & backend

13 minute read
, , , ,

Introduction

Hi, long time no see! In today’s post, I’ll present the architecture of the application I’m just starting to create — Minimalist. It’s a simple shopping list. I know it’s a basic project, but I’m building it to gain knowledge about a few lesser-known technologies and concepts. Another goal is to create a demo as quickly as possible, ideally in just a few evenings. It’s a random challenge for myself. In this post, I’ll also walk through the process of creating a simple backend demo.

Overview

The app will be built, as usual, using Flutter. However, there’s something new this time: I’ll be using what is probably the most popular state management framework—Flutter Bloc. Bloc is typically best suited for larger applications. This architecture helps keep a large codebase clean, simple, and predictable. My app will be small, but I plan to create a few blocs and hope to understand how it all works so I can build a larger project in the future.

The database will be MongoDB, a non-relational database system I’m familiar with. I enjoy using it in projects where the data model is less complex because it makes it easy to map documents to objects.

To create the API, I’ll use the Python framework FastAPI. While it might not be the most performant option for building backends, it allows for quick and efficient coding. The data model in MongoDB will be quite simple. The diagram looks as follows:

erd

For authentication and authorization, I’ll use the JWT Bearer flow from the OAuth 2 standards. FastAPI is also helpful here, providing many ready-made solutions. To implement product suggestion functionality, I’ll use Typesense, which is designed for this type of feature and serves as a lightweight, open-source alternative to Elasticsearch.

The entire development mini-backend will be deployed using Docker in the traditional way. The project will stay within a local environment, so there won’t be a need for systems like web servers, reverse proxies, load balancing, or caching.

OAuth2 - Bearer with JWT Tokens and Password Hashing

OAuth2 is an open set of standards aimed at building secure authorization mechanisms for different platforms. It is designed to grant specific access to particular resources. I will create a very simple mechanism where every user who logs in successfully will have access to all the endpoints used by the application. In larger projects, access to resources is usually granted in a more restrictive way, often using access groups. My application will use the Authorization Code Flow. This process involves three parties: the client, the resource server, and the authorization server. Thanks to the features of FastAPI, in my case, the resource owner and the authorization server are the same server.

When the client provides a valid username and password, it receives a pair of JSON Web Tokens (JWTs)—an access token and a refresh token. These are essentially encoded JSON objects containing information. They are not encrypted but are signed, which allows the server to verify that it is the token’s author. Obtaining these tokens for malicious purposes is not too difficult, which is why the access token usually has a short lifespan, such as a few minutes. Once it expires, the client must request a new token.

To ensure that the user does not get logged out frequently on the frontend, there is also a refresh token, which has a longer lifespan (e.g., several days) and is used to generate a new access token. When refreshing the token, it’s possible to generate a new refresh token and invalidate the old one, making both tokens’ lifespans shorter and the system more secure. If the refresh token becomes invalid, the user will be logged out on the frontend and will have to log in again to generate a new pair of tokens. FastAPI has a built-in set of procedures that are very helpful for implementing security.

In my case, the login flow will look as follows:

auth flow

Creating Models for Tokens and User

Let’s start by creating the models for tokens and the user.

class TokenModel(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str


class UserModel(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
    name: str = Field(...)
    email: str = Field(...)
    creationDate: datetime = Field(default_factory=datetime.utcnow)

    model_config = ConfigDict(
        populate_by_name=True,
        arbitrary_types_allowed=True,
    )

    @field_validator(
    "creationDate", mode="before"
    )
    @classmethod
    def check_none(cls, v, info):
        return generic_validator_function_with_default(cls, v, info)


class UserInDBModel(UserModel):
    hashedPassword: str


class CreateUserModel(UserModel):
    password: str


class LoginUserModel(BaseModel):
    email: str = Field(...)
    password: str = Field(...)

The inconsistency in field naming is due to the fact that the built-in OAuth2 methods in FastAPI require snake_case for token models. I use camelCase in the other models because that’s what I use on the frontend, making it simpler and faster. The user model has an alias in its id field because of MongoDB’s ID format; all models will use this alias. I also added a class method to set a default creationDate if it’s not provided.

The user model varies depending on the use case. When creating a user, we use the version with a plain text password, but when saving the user to the database, we use the hashed password model. For logging in, only the email and password are required, and for displaying the user information, the password is not needed.

Password Hashing and Verification

Now let’s create functions for hashing and verifying the user’s password using the passlib library and the bcrypt algorithm.

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)

We also create a function to authenticate the user using verify_password.

async def authenticate_user(email: str, password: str):
    user = await users.find_one({"email": email})

    if not user:
        return False

    if not verify_password(password, user["hashed_password"]):
        return False

    return user

Access Tokens

Next, we’ll handle access tokens. We generate two secret keys to sign both tokens using the following command:

openssl rand -hex 32

Store these keys and the algorithm (e.g., HS256) used for encoding in environment variables.

ACCESS_TOKEN_SECRET_KEY=f3855c768716c1ed7112d30c4cc6817ab8035abf195f18678679d4a1ba0acafc
REFRESH_TOKEN_SECRET_KEY=04be248833f68e793ae011203e415e5505b695284a45f8024ba35b0e80bdaef8
ALGORITHM=HS256

Create functions to generate and verify tokens.

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, ACCESS_TOKEN_SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


def create_refresh_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, REFRESH_TOKEN_SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


def verify_token(token: str, token_type: str):
    credential_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    expiration_exception = HTTPException(
        status_code=499 if token_type == "refresh" else 498,
        detail="Token expired",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        if token_type == "refresh":
            payload = jwt.decode(token, REFRESH_TOKEN_SECRET_KEY, algorithms=ALGORITHM)
        elif token_type == "access":
            payload = jwt.decode(token, ACCESS_TOKEN_SECRET_KEY, algorithms=ALGORITHM)

        user_id = payload.get("sub")

        if user_id is None:
            raise credential_exception

    except ExpiredSignatureError:
        raise expiration_exception
    except PyJWTError:
        raise credential_exception
    
    return user_id

I’ve created custom error codes for token expiration so that the client can easily decide whether to refresh the token or log out the user.

Endpoints

It’s time to move on to creating endpoints. I’m creating a separate APIRouter for the user-related endpoints.

user_router = APIRouter()

Later, it can be included in the application with the /users prefix as follows:

app.include_router(user_router, tags=["Users"], prefix="/users")

The first POST endpoint will be used to create a new user. It accepts the full user data. First, we check if a user with the given email address already exists; if so, we return a 400 error. Otherwise, the password is encrypted, a user model is created and inserted into the database, and we return a 200 - success.

async def create_user(user: CreateUserModel = Body(...)):
    existing_user = await users.find_one({"email": user.email})

    if existing_user:
        raise HTTPException(status_code=400, detail="E-mail already exists")

    encrypted_password = get_password_hash(user.password)

    user_for_db = UserInDBModel(
        name=user.name, email=user.email, hashed_password=encrypted_password
    ).model_dump(by_alias=True, exclude=["id"])

    result = await users.insert_one(user_for_db)

    return 200

The next POST endpoint is used for logging in. First, we try to log the user in by verifying the email and password. Then, we create a pair of refresh and access tokens, which are returned in the response. The client can save them and use them for authentication in subsequent sessions until the refresh token expires.

@user_router.post("/login", response_model=TokenModel)
async def login_for_access_tokens(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):

    user = await authenticate_user(form_data.username, form_data.password)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    refresh_token_expires = timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
    refresh_token = create_refresh_token(data={"sub": f"{str(user["_id"])}"}, expires_delta=refresh_token_expires)

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": f"{str(user["_id"])}"}, expires_delta=access_token_expires
    )

    token_model = TokenModel(access_token=access_token, refresh_token=refresh_token, token_type="bearer")

    return token_model

The next GET endpoint is for refreshing the access token. It accepts the refresh token and checks if it’s valid; if so, it generates a new access token. If not, it returns an appropriate error, and the client will need to prompt the user to log in again.

The last GET endpoint will be used to retrieve information about the currently logged-in user. It’s very simple, but the function it relies on is very important – all the subsequent endpoints in the application will depend on it. Let’s create it:

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    user_id = verify_token(token, "access")

    user = await users.find_one({"_id": ObjectId(user_id)})

    if user is None:
        raise HTTPException(detail="User does not exist")
    
    user_model = UserModel(id=str(user["_id"]), name=user["name"], email=user["email"], creation_date=user["creationDate"])

    return user_model

First, the access token is verified – if it’s missing, an appropriate error code is returned, and the client must attempt to refresh the token. Then, we attempt to retrieve the user object from the database and return the appropriate model. Now, let’s create an endpoint for retrieving user information:

@user_router.get('/me', response_model=UserModel, response_model_by_alias=False)
async def read_me(current_user: Annotated[UserModel, Depends(get_current_user)]):
    return current_user

That’s it! We have a complete flow for creating a user and logging in, as well as a method to secure additional endpoints in the application.

Items and Lists Endpoints

In this section, we will create routes with endpoints for operations related to lists and the items they contain. Let’s start with the Pydantic data models.

class ListModel(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
    name: str = Field(...)
    creatorId: PyObjectId = Field(...)
    creationDate: Optional[datetime] = Field(default_factory=datetime.utcnow)
    isShared: Optional[bool] = Field(default=False)
    participants: Optional[List[PyObjectId]] = Field(default=None)
    status: Optional[str] = Field(default="active")

    model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)

    @field_validator(
        "creationDate", "isShared", "status", mode="before"
    )
    @classmethod
    def check_none(cls, v, info):
        return generic_validator_function_with_default(cls, v, info)


class UpdateListModel(BaseModel):
    name: Optional[str] = None
    isShared: Optional[bool] = None
    participants: Optional[List[PyObjectId]] = None
    status: Optional[str] = None

    model_config = ConfigDict(
        arbitrary_types_allowed=True, json_encoders={ObjectId: str}
    )


class ListsCollection(BaseModel):
    lists: List[ListModel]

For shopping lists, we have three models. The main model includes all required fields:

  • id: A unique identifier automatically generated by MongoDB for the record in the collection.
  • name: The name of the shopping list.
  • creatorId: The identifier of the user who created the shopping list.
  • creationDate: The date the list was created, automatically generated server-side with the current date and time.
  • isShared: Indicates whether the list is shared; while the frontend for this project will not handle this, it might in the future.
  • participants: A list of users who are sharing the list.
  • status: The status of the list, for instance, if the frontend wants to allow for list archiving in the future.

We also have an update model, which only includes fields that can be updated, and a model containing a list of ListModel objects.

For items, we have three similar data models:

class ItemModel(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
    name: str = Field(...)
    description: Optional[str] = Field(default=None)
    creatorId: PyObjectId = Field(...)
    creationDate: Optional[datetime] = Field(default_factory=datetime.utcnow)
    isChecked: Optional[bool] = Field(default=False)
    listId: PyObjectId = Field(...)
    quantity: Optional[float] = Field(default=1)
    unit: Optional[str] = Field(default=None)
    price: Optional[float] = Field(default=None)

    model_config = ConfigDict(
        populate_by_name=True,
        arbitrary_types_allowed=True,
    )

    @field_validator(
    "creationDate", "isChecked", mode="before"
    )
    @classmethod
    def check_none(cls, v, info):
        return generic_validator_function_with_default(cls, v, info)


class UpdateItemModel(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    isChecked: Optional[bool] = None
    quantity: Optional[float] = None
    unit: Optional[str] = None
    price: Optional[float] = None

    model_config = ConfigDict(
        arbitrary_types_allowed=True, json_encoders={ObjectId: str}
    )


class ItemsCollection(BaseModel):
    items: List[ItemModel]

Attributes for an item include:

  • id: A unique identifier.
  • name: The name of the item.
  • description: A description of the item, optional.
  • creatorId: The identifier of the user who created the item.
  • creationDate: The date and time the item was created.
  • isChecked: The checked status of the item, defaults to False.
  • listId: The identifier of the list containing the item.
  • quantity: The quantity of the item, defaults to 1, optional.
  • unit: The unit of measurement for the quantity, optional.
  • price: The price of the item, optional.

Items also have separate models for updating and for returning lists of items.

CRUD

The route for lists contains endpoints for adding, updating, deleting, and retrieving all lists belonging to a specific user. All endpoints share a dependency: requiring a valid authorization token.

list_router = APIRouter(dependencies=[Depends(get_current_user)])


@list_router.post(
    "/",
    response_description="A new list added to the collection",
    response_model=ListModel,
    response_model_by_alias=False,
)
async def add_list(list_model: ListModel = Body(...)):
    list_dict = list_model.model_dump(by_alias=True, exclude=["id"], exclude_none=True)
    new_list = await lists.insert_one(list_dict)
    return list_dict


@list_router.get(
    "/{user_id}",
    response_description="Lists retrieved",
    response_model=ListsCollection,
    response_model_by_alias=False,
)
async def get_lists_by_user_id(user_id: str):
    lists_array = await lists.find({"creatorId": user_id}).to_list()
    return ListsCollection(lists=lists_array)


@list_router.put(
    "/{id}",
    response_description="List updated",
    response_model=ListModel,
    response_model_by_alias=False,
)
async def update_list(id: str, list_model: UpdateListModel = Body(...)):

    list_model = {
        k: v for k, v in list_model.model_dump(by_alias=True).items()
    }

    if len(list_model) >= 1:
        update_result = await lists.find_one_and_update(
            {"_id": ObjectId(id)},
            {"$set": list_model},
            return_document=ReturnDocument.AFTER,
        )
        if update_result is not None:
            return update_result
        else:
            raise HTTPException(status_code=404, detail=f"List {id} not found")

    if (existing_list_model := await lists.find_one({"_id": id})) is not None:
        return existing_list_model

    raise HTTPException(status_code=404, detail=f"List {id} not found")


@list_router.delete("/{id}", response_description="List removed")
async def remove_list(id: str):
    list_items = await items.find({"listId": id}).to_list()

    if list_items:
        for x in list_items:
            await items.find_one_and_delete({"_id": x['_id']})

    delete_list_result = await lists.find_one_and_delete({"_id": ObjectId(id)})

    if delete_list_result:
        return Response(status_code=status.HTTP_204_NO_CONTENT)

    raise HTTPException(
        status_code=404, detail=f"List {id} not found in the collection"
    )

The CRUD operations for items are quite similar:

@item_router.post(
    "/",
    response_description="A new item added to the collection",
    response_model=ItemModel,
    response_model_by_alias=False,
)
async def add_item(item_model: ItemModel = Body(...)):
    item_model = item_model.model_dump(by_alias=True, exclude=["id"])
    new_item = await items.insert_one(item_model)
    return item_model


@item_router.get(
    "/{list_id}",
    response_description="Items retrieved",
    response_model=ItemsCollection,
    response_model_by_alias=False,
)
async def get_item_by_list_id(list_id: str):
    items_array = await items.find({"listId": list_id}).sort("isChecked", 1).to_list()
    return ItemsCollection(items=items_array)


@item_router.put(
    "/{id}",
    response_description="Item updated",
    response_model=ItemModel,
    response_model_by_alias=False,
)
async def update_item(id: str, item_model: UpdateItemModel = Body(...)):

    item_model = {
        k: v for k, v in item_model.model_dump(by_alias=True).items()
    }

    if len(item_model) >= 1:
        update_result = await items.find_one_and_update(
            {"_id": ObjectId(id)},
            {"$set": item_model},
            return_document=ReturnDocument.AFTER,
        )
        if update_result is not None:
            return update_result
        else:
            raise HTTPException(status_code=404, detail=f"Item {id} not found")

    if (existing_item_model := await items.find_one({"_id": id})) is not None:
        return existing_item_model

    raise HTTPException(status_code=404, detail=f"Item {id} not found")


@item_router.delete("/{id}", response_description="Item removed")
async def remove_item(id: str):
    delete_result = await items.find_one_and_delete({"_id": ObjectId(id)})

    if delete_result:
        return Response(status_code=status.HTTP_204_NO_CONTENT)

    raise HTTPException(
        status_code=404, detail=f"item {id} not found in the collection"
    )

MongoDB Connection

As you can see, almost all endpoints interact with the database. Setting up the connection is straightforward. For asynchronous MongoDB operations, I use the Motor driver, connecting to all necessary collections:

mongo_client = motor.motor_asyncio.AsyncIOMotorClient(DB_URL)

db = mongo_client["minimalist"]

users = db["users"]

users_settings = db["users_settings"]

lists = db["lists"]

items = db["items"]

Product Suggestions

Product suggestions are displayed as users type phrases to add items to their shopping lists. Typesense, an open-source full-text search engine, is an excellent alternative to expensive or complex solutions like Elasticsearch.

The implementation is straightforward. In the same file where I connect to MongoDB, I add the connection to Typesense (for simplicity, this is a basic local, development-only setup):

typesense_client = typesense.Client({
    'nodes': [{
        'host': 'localhost', 
        'port': '8108',
        'protocol': 'http'
    }],
    'api_key': 'xyz',
    'connection_timeout_seconds': 2
})


collection_schema = {
    'name': 'products',
    'fields': [
        {'name': 'name', 'type': 'string'},
    ]
}

try:
    typesense_client.collections.create(collection_schema)
except typesense.exceptions.ObjectAlreadyExists:
    pass

Next, we populate the suggestion database with products that will match user queries.

products = [
    {"name": "Water"},
    {"name": "Milk"},
    {"name": "Cheese"},
    # ... Add more items
]

for product in products:
    try:
        response = typesense_client.collections['products'].documents.create(product)
        print(f"Product added: {product['name']}")
    except Exception as e:
        print(f"Failed to add product {product['name']}: {e}")

Finally, we add an endpoint to query suggestions as users type:

@item_router.get("/suggest/")
async def suggest_products(query: str):
    try:
        search_parameters = {
            'q': query,
            "query_by": 'name',
            'prefix': True,
            'limit': 5
        }
        
        results = typesense_client.collections['products'].documents.search(search_parameters)
        return [hit["document"]["name"] for hit in results["hits"]]
    except Exception as e:
        print(e)
        raise HTTPException(status_code=500)

If I were developing the application for production, I would enhance the suggestion feature by adding frequently searched phrases entered by the app’s users to the collection.

This improvement would not only make the suggestion engine more dynamic and user-centric but would also help in identifying trends and patterns in user behavior. By storing and analyzing these commonly entered phrases, the application could prioritize popular or trending items in the suggestions, improving the overall user experience.

Moreover, combining this with analytics could offer valuable insights into user preferences, enabling better personalization and even informing future product or feature decisions.

Docker

To run the API and required services, I create a Dockerfile and a Docker Compose configuration:

FROM python:3.14-slim-bookworm

WORKDIR /code

COPY ./requirements.txt ./requirements.txt

RUN pip install --no-cache-dir --upgrade -r ./requirements.txt

COPY ./app/ .

ENTRYPOINT ["python", "main.py"]

version: "3"

services:
  minimalist-api:
    build: ./api
    ports:
      - 8000:8000
    depends_on:
      - "mongodb"
      - "typesense"
    networks:
      - minimalist-network

  mongodb:
    image: mongo
    ports:
      - 27017:27017
    networks:
      - minimalist-network

  typesense:
    image: typesense/typesense:27.1
    ports:
      - "8108:8108"
    networks:
      - minimalist-network

networks:
  minimalist-network:
    name: minimalist-network

Once the services are running, all created endpoints are visible at /docs.

api

Conclusion

Building a basic, development-ready API with FastAPI is relatively simple. The library facilitates creating systems that include seemingly complex features like JWT-based authentication and authorization. For production, you’d need to secure all database services, think about scaling, and implement data backup and recovery plans—making the task significantly more challenging.

In the next post, I’ll create a frontend using Flutter. Cya!