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 Flaskfrom flask_restx import Apifrom 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 jsonimport osfrom collections import UserListfrom json import JSONEncoderfrom sqlalchemy import Column, ForeignKey, Integer, Tablefrom sqlalchemy.ext.compiler import compilesfrom sqlalchemy.orm import relationshipfrom sqlalchemy.sql import Insertfrom sqlalchemy_serializer import SerializerMixinfrom 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 stmtInsert.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_requestdef 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!…