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!…
English