ML App P2 – Flask with SqlAlchemy, SqLite and Docker

This is the second part of the serie of posts ML App – Flask with SqlAlchemy, SqLite and Docker
For this app we gonna use the technologies:

• Python Flask – Python backend dev and web api;
• SqlAlchemy – SQL toolkit and Object Relational Mapper that gives application developers the full power and flexibility of SQL;
• SqLite – Database engine to store data;
• Docker – Plattaform to create containers to launch our systems on cloud servers.

Steps:

1 – Database: We gonna use SqlAlchemy to map our classes and create our database with their respectives tables on SqLite.

Create a folder to store locally our project and create inside the file __init__.py with the content:

from flask import Flask
from flask_restx import Api
from flask_sqlalchemy import SQLAlchemy

"""flask app wrapper for Flask
"""
app = Flask(__name__)

"""Configuration for Flask database
"""
DIR_DB = "movies.db"
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{DIR_DB}"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True

"""Creation of the database for Flask
"""
db = SQLAlchemy(app)

1.2 – Let’s create our models.py

import json
import os
from collections import UserList
from json import JSONEncoder

from sqlalchemy import Column, ForeignKey, Integer, Table
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import relationship
from sqlalchemy.sql import Insert
from sqlalchemy_serializer import SerializerMixin

from app import app, db


@compiles(Insert, "sqlite")
def suffix_insert(insert, compiler, **kwargs):
    stmt = compiler.visit_insert(insert, **kwargs)
    if insert.dialect_kwargs.get("sqlite_on_conflict_do_nothing"):
        stmt += " ON CONFLICT DO NOTHING"
    return stmt


Insert.argument_for("sqlite", "on_conflict_do_nothing", True)


class Base(db.Model, SerializerMixin):
    """Base class for other models

    Args:
        db (_type_): _description_
        SerializerMixin (_type_): SerializerMixin
    """

    __abstract__ = True

    id = db.Column(db.Integer, primary_key=True)
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
    updated_at = db.Column(
        db.DateTime,
        default=db.func.current_timestamp(),
        onupdate=db.func.current_timestamp(),
    )


class Type(Base):
    """_summary_

    Args:
        Base (_type_): _description_
    """

    __tablename__ = "type"

    # serialize_only = ("name", "description", "values")
    # serialize_rules = ()
    
    def __init__(self, name=None, description=None):
        self.name = name
        self.description = description        

    def __setattr__(self, __name, __value) -> None:
        super().__setattr__(__name, __value)
    
    name = db.Column(db.String(128), unique=True)
    description = db.Column(db.Text)

    # one-to-many collection
    titles = db.relationship("Title", backref="type")
    db.UniqueConstraint(name)


genre_title = Table(
    "genre_title",
    Base.metadata,
    Column("title_id", ForeignKey("title.id"), primary_key=True),
    Column("genre_id", ForeignKey("genre.id"), primary_key=True),
)


class Genre(Base):
    """_summary_

    Args:
        Base (_type_): _description_
    """

    def __init__(self, name=None):
        self.name = name

    name = db.Column(db.String(128), unique=True)


class Title(Base):
    """_summary_

    Args:
        Base (_type_): _description_
    """

    __tablename__ = "title"

    # serialize_only = ("primaryTitle", "primaryTitle")
    # serialize_rules = ()

    def __init__(
        self,
        tconst=None,
        titletype=None,
        primarytitle=None,
        originaltitle=None,
        isadult=None,
        startyear=None,
        endyear=None,
        runtimeminutes=None,
        genres=None,
        type=None
    ):
        self.tconst = tconst
        self.primary_title = primarytitle
        self.original_title = originaltitle
        self.is_adult = isadult
        self.start_year = startyear
        self.end_year = endyear
        self.runtime_minutes = runtimeminutes
        self.titletype = titletype
        self.genres = genres,    
        self.type = type   

    tconst = db.Column(db.String(256))
    primary_title = db.Column(db.String(256))
    original_title = db.Column(db.String(256))
    is_adult = db.Column(db.String(256))
    start_year = db.Column(db.String(256))
    end_year = db.Column(db.String(256))
    runtime_minutes = db.Column(db.String(256))

    type_id = Column(db.Integer, db.ForeignKey("type.id"))
    lst_genres = relationship("Genre", secondary=genre_title)

So far we have the following diagram:

2 – Let’s create our api.py to hadle our database with the operations necessaries:

2.1 the first method gonna create our database and dables according with our models.py:

@app.before_first_request
def before_first_request():
    """Before first request let's create the database and the structures.

    Returns:
        _type_: _description_
    """
    db.drop_all()
    db.session.commit()
    db.create_all()
    return "All the database is truncated!"

So til this moment we create the structure of our app and the database necessary to storage our data.

See you on the next post!…