UD - FLASK - SQLAlchemy

https://arac.tecladocode.com/

• No manually create table

• No manually build connection & query as vanilla sql.

1. Init

# db.py
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

# app.py
from flask import Flask
...

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///data.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False # turn-off flaskSqlalchemy track, not sqlalchemy track
app.config['PROPAGATE_EXCEPTIONS'] = True
app.secret_key = 'jose'
api = Api(app)


@app.before_first_request # automatic db creation
def create_tables():
    db.create_all()

...

if __name__ == '__main__':
    from db import db
    db.init_app(app)
    app.run(port=5000, debug=True)

2. Model

# models/item.py
from db import db


class ItemModel(db.Model):
    __tablename__ = 'items'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80))
    price = db.Column(db.Float(precision=2))

    store_id = db.Column(db.Integer, db.ForeignKey('stores.id')) # relation
    store = db.relationship('StoreModel')                        # relation

    def __init__(self, name, price, store_id):
        self.name = name
        self.price = price
        self.store_id = store_id                                # relation

    def json(self):
        return {'name': self.name, 'price': self.price}

    @classmethod
    def find_by_name(cls, name):
        return cls.query.filter_by(name=name).first()

    def save_to_db(self):
        db.session.add(self)
        db.session.commit()

    def delete_from_db(self):
        db.session.delete(self)
        db.session.commit()

# models/store.py
from db import db


class StoreModel(db.Model):
    __tablename__ = 'stores'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80))

    items = db.relationship('ItemModel', lazy='dynamic')    # relation

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

    def json(self):
        return {'name': self.name, 'items': [item.json() for item in self.items.all()]}

    @classmethod
    def find_by_name(cls, name):
        return cls.query.filter_by(name=name).first()

    def save_to_db(self):
        db.session.add(self)
        db.session.commit()

    def delete_from_db(self):
        db.session.delete(self)
        db.session.commit()

3. Controller

# resources/item.py
from flask_restful import Resource, reqparse
from flask_jwt import jwt_required
from models.item import ItemModel


class Item(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('price',
                        type=float,
                        required=True,
                        help="This field cannot be left blank!"
                        )
    parser.add_argument('store_id',
                        type=int,
                        required=True,
                        help="Every item needs a store_id."
                        )

    @jwt_required()
    def get(self, name):
        item = ItemModel.find_by_name(name)
        if item:
            return item.json()
        return {'message': 'Item not found'}, 404

    def post(self, name):
        if ItemModel.find_by_name(name):
            return {'message': "An item with name '{}' already exists.".format(name)}, 400

        data = Item.parser.parse_args()

        item = ItemModel(name, **data)

        try:
            item.save_to_db()
        except:
            return {"message": "An error occurred inserting the item."}, 500

        return item.json(), 201

    def delete(self, name):
        item = ItemModel.find_by_name(name)
        if item:
            item.delete_from_db()
            return {'message': 'Item deleted.'}
        return {'message': 'Item not found.'}, 404

    def put(self, name):
        data = Item.parser.parse_args()

        item = ItemModel.find_by_name(name)

        if item:
            item.price = data['price']
        else:
            item = ItemModel(name, **data)

        item.save_to_db()

        return item.json()


class ItemList(Resource):
    def get(self):
        return {'items': list(map(lambda x: x.json(), ItemModel.query.all()))} 
        # ItemModel.query.all can be moved to models

4. Relationship

store has many items

• models

# models/item.py
from db import db

class ItemModel(db.Model):
    __tablename__ = 'items'

    ...
    store_id = db.Column(db.Integer, db.ForeignKey('stores.id'))
    store = db.relationship('StoreModel')

    def __init__(self, name, price, store_id):
        ...
        self.store_id = store_id

# models/store.py
from db import db

class StoreModel(db.Model):
    __tablename__ = 'stores'

    ...
    items = db.relationship('ItemModel', lazy='dynamic') # lazy: fast in create a store

    def __init__(self, name):
        ...

    def json(self):
        return {'name': self.name, 'items': [item.json() for item in self.items.all()]} 
        # store.json() includes all items.json()
        # self.items is now only a query since lazy
        # self.items.all() execute query & take long time

• controllers

# resources/item.py
class Item(Resource):
    parser = reqparse.RequestParser()
    ...
    parser.add_argument('store_id',
        type=int,
        required=True,
        help="Every item needs a store_id."
    )

    def post(self, name):
        ...
        data = Item.parser.parse_args()
        item = ItemModel(name, **data) # data['store_id']

        try:
            item.save_to_db()
        ...

5. FLASK-SQLAlchemy & SQLAlchemy

You will have the db.session and you can use it as if it was pure SQLAlchemy setup.

engine: starting point of SQLAlchemy

engine + connection: RAW SQL query

session: (1) db proxy, (2) ORM query

• SQLAlchemy

from flask import Flask
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

app = Flask(__name__)

engine = create_engine('sqlite:///foo.db')

Base = declarative_base()           # create base class which stores the classes representing tables
Base.metadata.reflect(bind=engine)

Session = sessionmaker(bind=engine) # create a configured "Session" class
session = Session()                 # create a Session instance

Base.metadata.create_all(engine)    # create all tables that don't yet exist

• FLASK-SQLAlchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///foo.db'
db = SQLAlchemy(app) # Magic: creates everything; Base (db.Model), engine (db.engine) & session (db.session)

# create all tables
from app import models
models.db.create_all()