My Journey to Help Build a P2P Social Network – Database Code Structure

Welcome to part 3 of this series, and thank you for sticking around!

Iโ€™ve come to realize this might become a rather long series. The main reason for that is that it documents two things. This is the birth of an application and my personal journey in developing that application. I know parts 1 and 2 have been very wordy. This will change now. I promise that you will see a lot of code in this episode :-).

Database Code

So after that slight philosophical tidbit, it is time to dive into the actual database code. As I mentioned in the previous article, I chose to use Deta Space as the database provider. There are two reasons for this. The first is the ease of use and the second are its similarities to my favorite NoSQL database MongoDB.

๐Ÿ’ก Recommended: Please check my article on creating a shopping list in Streamlit on how to set it up. It takes only a few minutes.

For reference, the server directory with all the code to get the FastAPI server working looks as follows:

โ”œโ”€โ”€ requirements.txt
โ”œโ”€โ”€ .env

All the database code will live in the file. For the Pydantic models, Iโ€™ll use

Database Functions

The database functions are roughly divided into three parts.

  • We first need functionality for everything related to users.
  • Next, we need to code that handles everything related to adding and managing friends.
  • The third part applies to all the code for managing thoughts. Thoughts are Peerbrainโ€™s equal of messages/tweets.

The file will also contain some helper functions to aid with managing the public keys of users. Iโ€™ll go into a lot more detail on this in the article about encryption.

To set up our file we first need to import everything needed. As before, Iโ€™ll show the entire list and then explain what everything does once we write code that uses it.

"""This file will contain all the database logic for our server module. It will leverage the Deta Base NoSQL database api."""

from datetime import datetime
import math
from typing import Union
import os
import logging
from pprint import pprint #pylint: disable=unused-import
from uuid import uuid4
from deta import Deta
from dotenv import load_dotenv
from passlib.context import CryptContext

The #pylint comment you can see above is used to make sure pylint skips this import. I use pprint for displaying dictionaries in a readable way when testing. As I donโ€™t use it anywhere in the actual code, pylint would start to fuss otherwise.

๐Ÿ’ก Tip: For those interested, pylint is a great tool to check your code for consistency, errors and, code style. It is static, so it canโ€™t detect errors occurring at runtime. I like it even so๐Ÿ™‚.

After having imported everything, I first initialize the database. The load_dotenv() below, first will load all my environment variables from the .env file. 


#---DB INIT---#
DETA_KEY = os.getenv("DETA_KEY")
deta = Deta(DETA_KEY)
USERS = deta.Base("users")
THOUGHTS = deta.Base("thoughts")
KEYS = deta.Base("keys_db")

Once the variables are accessible, I can use the Deta API key to initialize Deta. Creating Bases in Deta is as easy as defining them with deta.Base. I can now call the variable names to perform CRUD operations when needed.

Generate Password Hash

The next part is very important. It will generate our password hash so the password is never readable. Even if someone has control of the database itself, they will not be able to use it. Cryptcontext itself is part of the passlib library. This library can hash passwords in multiple ways.๐Ÿ™‚.

pwd_context = CryptContext(schemes =["bcrypt"], deprecated="auto")
def gen_pw_hash(pw:str)->str:
    """Function that will use the CryptContext module to generate and return a hashed version of our password"""
    return pwd_context.hash(pw)

User Functions

The first function of the user functions is the easiest. It uses Detaโ€™s fetch method to retrieve all objects from a certain Base, deta.users, in our case.

def get_users() -> dict:
    """Function to return all users from our database"""
        return {user["username"]: user for user in USERS.fetch().items}
    except Exception as e:
        # Log the error or handle it appropriately
        print(f"Error fetching users: {e}")
        return {}

The fact that the function returns the found users as a dictionary makes them easy to use with FastAPI. As we contact a database in this function and all the others in this block, a tryexcept block is necessary. 

The next two functions are doing the same thing but with different parameters. They accept either a username or an email.

I am aware that these two could be combined into a single function with an if-statement. I still do prefer the two separate functions, as I find them easier to use. Another argument I will make is also that the email search function is primarily an end user function. I plan to use searching by username in the background as a helper function for other functionality.

def get_user_by_username(username:str)->Union[dict, None]:
    """Function that returns a User object if it is in the database. If not it returns a JSON object with the
    message no user exists for that username"""
        if (USERS.fetch({"username" : username}).items) == []:
            return {"Username" : "No user with username found"}
            return USERS.fetch({"username" : username}).items[0]
    except Exception as error_message:
        return None

def get_user_by_email(email:str)->Union[dict, None]:
    """Function that returns a User object if it is in the database. If not it returns a JSON object with the
    message no user exists for that email address"""
        if (USERS.fetch({"email" : email}).items) == []:
            return {"Email" : "No user with email found"}
            return USERS.fetch({"email" : email}).items[0]
    except Exception as error_message:
        return None

The functions above both take a parameter that they use to filter the fetch request to the Deta Base users.

If that filtering results in an empty list a proper message is returned. If the returned list is not empty, we use the .items method on the fetch object and return the first item of that list. In both cases, this will be the user object that contains the query string (email or username).

The entire sequence is run inside a try-except block as we are trying to contact a database.

Reset User Password

When working with user creation and databases, a function to reset a userโ€™s password is required. The next function will take care of that.

def change_password(username, pw_to_hash):
    """Function that takes a username and a password in plaintext. It will then hash that password>
    After that it creates a dictionary and tries to match the username to users in the database. If
    successful it overwrites the previous password hash. If not it returns a JSON message stating no
    user could be found for the username provided."""
    hashed_pw = gen_pw_hash(pw_to_hash)
    update= {"hashed_pw": hashed_pw }
        user = get_user_by_username(username)
        user_key = user["key"]
        if not username in get_users():
            return {"Username" : "Not Found"}
            return USERS.update(update, user_key), f"User {username} password changed!"
    except Exception as error_message:
        return None  

This function will take a username and a new password. It will first hash that password and then create a dictionary. Updates to a Deta Base are always performed by calling the update method with a dictionary. As in the previous functions, we always check if the username in question exists before calling the update. Also, donโ€™t forget the try-except block!

Create User

The last function is our most important one :-). You canโ€™t perform any operations on user objects if you have no way to create them! Take a look below to check out how weโ€™ll handle that.

def create_user(username:str, email:str, pw_to_hash:str)->None:
    """Function to create a new user. It takes three strings and inputs these into the new_user dictionary. The function then
    attempts to put this dictionary in the database"""

    new_user = {"username" : username,
                "key" : str(uuid4()),
                "hashed_pw" : gen_pw_hash(pw_to_hash),
                "email" : email,
                "friends" : [],
                "disabled" : False}
        return USERS.put(new_user)
    except Exception as error_message:
        return None

The user creation function will take a username, email, and password for now. It will probably become more complex in the future, but it serves our purposes for now. Like the Deta update method, creating a new item in the database requires a dictionary. Some of the necessary attributes for the dictionary are generated inside the function.

The key needs to be unique, so we use Pythonโ€™s uuid4 module. The friendโ€™s attribute will contain the usernames of other users but starts as an empty list. The disabled attribute, finally, is set to false. 

After finishing the initialization, creating the object is a matter of calling the Deta put method. I hear some of you thinking that we donโ€™t do any checks if the username or email already exists in the database. You are right, but I will perform these checks on the endpoint receiving the post request for user creation.

Some Coding Thoughts and Learnings

GitHub ๐Ÿ‘ˆ Join the open-source PeerBrain development community!

One thing that never ceases to amaze me is the amount of documentation I like to add. I do this first in the form of docstrings as it helps me keep track of what function does what. I find it boring most of the time, but in the end, it helps a lot!

The other part of documenting that I like is type hints. I admit they sometimes confuse me still, but I can see the merit they have when an application keeps growing. 

We will handle the rest of the database function in the next article. See you there!

Participate in Building the Decentralized Social Brain Network ๐Ÿ‘‡

As before, I state that I am completely self-taught. This means Iโ€™ll make mistakes. If you spot them, please post them on Discord so I can remedy them ๐Ÿ™‚. 

As always, feel free to ask me questions or pass suggestions! And check out the GitHub repository for participation!

๐Ÿ‘‰ GitHub: