Flask SQLAlchemy

---

1. Basic Model Definition:

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

2. Relationships (One-to-Many):

class Order(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    user = db.relationship('User', backref=db.backref('orders', lazy=True))

3. Relationships (Many-to-Many):

users_tags = db.Table('users_tags',
    db.Column('user_id', db.Integer, db.ForeignKey('user.id')),
    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'))
)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    tags = db.relationship('Tag', secondary=users_tags, backref=db.backref('users', lazy=True))

class Tag(db.Model):
    id = db.Column(db.Integer, primary_key=True)

4. Querying with Filters:

users = User.query.filter(User.username.like('%user%')).all()

5. Querying with Ordering:

users = User.query.order_by(User.username.desc()).all()

6. Querying with Pagination:

users = User.query.paginate(page=1, per_page=10)

7. Limiting Query Results:

users = User.query.limit(10)

8. Deleting Records:

db.session.delete(user)
db.session.commit()

9. Updating Records:

user.username = 'new_username'
db.session.commit()

10. Creating a Foreign Key:

class Order(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

11. Adding a Timestamp:

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    created_at = db.Column(db.DateTime, default=datetime.datetime.now)

12. Using a Model Mixin:

from sqlalchemy.orm import validates

class ValidatedMixin:
    @validates('name')
    def validate_name(self, key, value):
        if len(value) < 5:
            raise ValueError('Name must be at least 5 characters long')
        return value

class User(db.Model, ValidatedMixin):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)

13. Creating a Composite Primary Key:

class Track(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    artist = db.Column(db.String(80), nullable=False)
    title = db.Column(db.String(80), nullable=False)
    __table_args__ = (db.PrimaryKeyConstraint('artist', 'title'),)

14. Defining Computed Columns:

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    first_name = db.Column(db.String(80), nullable=False)
    last_name = db.Column(db.String(80), nullable=False)
    full_name = db.computed("first_name + ' ' + last_name", persisted=True)

15. Using a Model as a Mixin:

class TimeStampedMixin:
    created_at = db.Column(db.DateTime, default=datetime.datetime.now)
    updated_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

class User(db.Model, TimeStampedMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), nullable=False)

16. Using a Model as a Data Transfer Object (DTO):

from marshmallow import Schema, fields

class UserSchema(Schema):
    id = fields.Int()
    username = fields.Str()

user_schema = UserSchema()
user_json = user_schema.dump(user)

17. Serializing Models to JSON:

from flask import jsonify

@app.route('/users')
def get_users():
    users = User.query.all()
    return jsonify([user_schema.dump(user).data for user in users])

18. Deserializing JSON to Models:

from flask import request, jsonify

@app.route('/users', methods=['POST'])
def create_user():
    user_data = request.get_json()
    user = UserSchema().load(user_data).data
    db.session.add(user)
    db.session.commit()
    return jsonify(user_schema.dump(user).data)

19. Using a Database URL:

app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://user:password@host:port/database'

20. Connecting to a Database:

db.init_app(app)

21. Creating Database Tables:

db.create_all()

22. Dropping Database Tables:

db.drop_all()

23. Using a Custom Model Manager:

from flask_sqlalchemy import Model

class User(Model):

    def __init__(self, *args, **kwargs):
        Model.__init__(self, *args, **kwargs)
        self._events = []

    def add_event(self, event):
        self._events.append(event)

24. Customizing Model Event Handling:

@event.listens_for(User, 'before_commit')
def before_commit_user(mapper, connection, target):
    target._events.append('before_commit')

@event.listens_for(User, 'after_commit')
def after_commit_user(mapper, connection, target):
    target._events.append('after_commit')

25. Using a Database Proxy:

app.config['SQLALCHEMY_DATABASE_URI_PROXY'] = 'host:port'

26. Using a Database Engine:

from sqlalchemy import create_engine

engine = create_engine('postgresql://user:password@host:port/database')

27. Using a Session Factory:

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

28. Using a Scoped Session:

from flask_sqlalchemy import scoped_session

db_session = scoped_session(Session, scopefunc=lambda: request)

29. Using a Transaction:

with session.begin():
    # Do something

30. Using a Rollback:

try:
    with session.begin():
        # Do something
        session.rollback()
except Exception:
    session.rollback()

31. Using a Commit:

with session.begin():
    # Do something
    session.commit()

32. Using a Flush:

session.flush()

33. Using a Query:

query = session.query(User)

34. Using a Filter:

query = query.filter(User.username == 'user')

35. Using an Order By:

query = query.order_by(User.username)

36. Using a Limit:

query = query.limit(10)