Peewee ORM Integration

Installation

pip install peewee
pip install mysqlclient
pip install pymysql

In the root of the project you must create a file 'database.py':

from peewee import *
database = MySQLDatabase(
    'fastapidb',
    host='localhost',
    port=3306,
    user='root',
    passwd=''
)

Then we import the connection in the 'main.py' file, in addition to this we use the events to establish the connection and close the connection:

from fastapi import FastAPI
from database import database as connection

app = FastAPI(title='The title',
              description='The description',
              version=1)

# Events

@app.on_event('startup')
def startup():
    if connection.is_closed():
        connection.connect()
        print('Connecting the DataBase...')

@app.on_event('shutdown')
def shutdown():
    if not connection.is_closed():
        connection.close()
        print('Connection closed')

Models

The "models" directory is created and the '.py' files for each model are stored inside:

from peewee import *
from database import database
from datetime import datetime

class User(Model):
    username = CharField(max_length=50, unique=True, index=True)
    email = CharField(max_length=60, null=False)
    active = BooleanField(default=False)
    created_at = DateTimeField(default=datetime.now)

    @classmethod
    def hash_password(cls, password):
        password_hash = hashlib.md5()
        password_hash.update(password.encode('utf-8'))
        return password_hash.hexdigest()

    def __str__(self):
        return self.username

    class Meta:
        database = database
        db_table = 'users'

To create the tables automatically when starting the server, in the 'main.py' file the models are imported and executed as follows:

from fastapi import FastAPI
from database import database as connection
from models.user import User
from models.article import Article

app = FastAPI(title='The title',
              description='The description',
              version=1)

# Events

@app.on_event('startup')
def startup():
    if connection.is_closed():
        connection.connect()
    connection.create_tables([User, Article])

Insert a record

1. If any validation exists, a class is created in the 'validations' directory

from fastapi import HTTPException
from pydantic import BaseModel, validator
from models.user import User

class UserValidationModel(BaseModel):
    username: str
    email: str
    password: str

@validator('username')
def username_len_validator(cls, username):
    if len(username) < 3> 50:
        raise ValueError('The username must have between 3 and 50 characters')
    return username

@validator('username')
def username_available(cls, username):
    if User.select().where(User.username == username).exists():
        return HTTPException(409, 'The username is not available')
    return username

2. The validation class is imported in 'main.py'

from validations.createUser import UserValidationModel

3. The route is created in 'main.py' and the function is implemented

@ app.post('/users/')
async def create_user(user: UserValidationModel):

if isinstance(user.username, HTTPException):
    return user.username
password = User.hash_password(user.password)
user = User.create(
    username=user.username,
    email=user.email,
    password=password
)
return user.id

Thanks for reading :)
I invite you to continue reading other entries and visiting us again soon.