1. Define tables:

from sqlalchemy import create_engine, inspect, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 1. Define tables:
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(length=64))
    age = Column(Integer)

class Address(Base):
    __tablename__ = 'address'

    id = Column(Integer, primary_key=True)
    address = Column(String(length=64))
    # A `FOREIGN KEY` is a field (or collection of fields) in one table, that refers to the `[PRIMARY KEY]` in another table
    user_id = Column(Integer, ForeignKey("users.id")) # 字段user_id是外键，依赖users.id

    user = relationship("User") # 通过orm查address表的时候，user也会被查询一起带出来

2. Get an Engine, which the Session(a session factory below) will use for connection resources:

# 2. Get an Engine, which the Session will use for connection resources:
# create_engine 返回一个 engine实例，实例只是保存了一些属性/配置，还没有实际去连接db. 
engine = create_engine('mysql+pymysql://user:password@dbhost/mydatabase')

# # engine.connect() returns a connection, which can run raw sql
# conn = engine.connect()
# res = conn.execute("select * from users limit 10")
# print(type(res)) # <class 'sqlalchemy.engine.cursor.LegacyCursorResult'>
# for row in res:
#     print("name", row["name"]) #  row["name"] can be replaced with `row.name`
# conn.close() # after close()，the underlying DBAPI connection is then returned to the connection pool, which is referenced by this engine

3. Create tables and reflect tables:

# 3. Create tables and reflect tables:
Base.metadata.create_all(engine)

inspector = inspect(engine)
# 获取表的metadata
print(inspector.get_columns('users'))

4. Get a session factory:

# 4. Get a session factory:
Session = sessionmaker(bind=engine) # Session is a sessionmaker instance

5. read from db table in ORM mode:

# 5. read from db table in ORM mode
session = Session() # this invokes Session.__call__() and get a instance of <class 'sqlalchemy.orm.session.Session'>

# session.query() returns a new object of :class:`_query.Query` corresponding to this session`
# usage: session.query(User, Address)
users : list[User] = session.query(User).filter(User.age > 25).all()
for user in users:
    # here, `user`'s type is `User`, because param `entities` passed to query() is `User`
    print(user.name)

age = User.age.label("age_alias") # 起一个select别名
cursor_results = session.query(User.name, age).filter(User.age > 10).all()
for user in cursor_results:
    # here, `user`'s type is <class 'sqlalchemy.engine.row.Row'>, because param `entities` passed to query() is customed columns
    print(user.name, user.age_alias)

with Session() as s:
    res : list[Address] = s.query(Address).all()
    for r in res:
        print(r.address, r.user.name)

6. write to db table in ORM mode:

# 6. write to db table in ORM mode
with Session() as session:
    session.add(some_object)
    session.add(some_other_object)
    session.commit()

---

total code:

from sqlalchemy import create_engine, inspect, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# 1. Define tables:
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

class Address(Base):
    __tablename__ = 'address'

    id = Column(Integer, primary_key=True)
    address = Column(String(length=64))
    user_id = Column(Integer, ForeignKey("users.id")) # 字段user_id是外键，依赖users.id

    user = relationship("User") # 通过orm查address表的时候，user也会被查询一起带出来

# 2. Get an Engine, which the Session will use for connection resources:
# create_engine 返回一个 engine实例，实例只是保存了一些属性/配置，还没有实际去连接db. 
engine = create_engine('mysql+pymysql://user:password@dbhost/mydatabase')

# # engine.connect() returns a connection, which can run raw sql
# conn = engine.connect()
# res = conn.execute("select * from users limit 10")
# print(type(res)) # <class 'sqlalchemy.engine.cursor.LegacyCursorResult'>
# for row in res:
#     print("name", row["name"]) #  row["name"] can be replaced with `row.name`
# conn.close() # after close()，the underlying DBAPI connection is then returned to the connection pool, which is referenced by this engine

# 3. Create tables and reflect tables:
Base.metadata.create_all(engine)

inspector = inspect(engine)
# 获取表的metadata
print(inspector.get_columns('users'))

# 4. Get a session factory:
Session = sessionmaker(bind=engine) # Session is a sessionmaker instance

# 5. read from db in ORM mode
session = Session() # this invokes sessionmaker.__call__() and get a instance of <class 'sqlalchemy.orm.session.Session'>

# session.query() returns a new object of :class:`_query.Query` corresponding to this session`
# usage: session.query(User, Address)
users : list[User] = session.query(User).filter(User.age > 25).all()
for user in users:
    # here, `user`'s type is `User`, because param `entities` passed to query() is `User`
    print(user.name)

age = User.age.label("age_alias") # 起一个select别名
cursor_results = session.query(User.name, age).filter(User.age > 10).all()
for user in cursor_results:
    # here, `user`'s type is <class 'sqlalchemy.engine.row.Row'>, because param `entities` passed to query() is customed columns
    print(user.name, user.age_alias)

with Session() as s:
    res : list[Address] = s.query(Address).all()
    for r in res:
        print(r.address, r.user.name)


# 6. write to db in ORM mode
with Session() as session:
    session.add(some_object)
    session.add(some_other_object)
    session.commit()

refer to
https://github.com/Realmvls/note-1/blob/master/note/python/sqlalchemy.md

sqlalchemy使用进阶 - building dynamic filter

我们在读写db table的时候，经常需要组合各种各样的过滤条件，于是我们希望把【组合过滤条件】的能力封装到一个独立的小方法中。这里以上文的users表为例，给出两种比较优雅的building dynamic filter方式（推荐第二种）

from pydantic import BaseModel

class UserFilter(BaseModel):
    """
    table 'users' 支持的查询条件
    """
    id : int = None
    name : str = None
    age : int = None

def build_query_with_dynamic_filter(session, cond: UserFilter):
    """
    a helper func, 根据 cond 返回一个 orm Query with dynamic filter

    这种写法利用or_实现“自动把非None字段作为过滤条件”，缺点是当增减查询条件时，需要手动增减filter的内容
    """
    return session.query(User) \
        .filter(
            or_(User.id == cond.id, cond.id is None),
            or_(User.age == cond.age, cond.age is None),
            or_(User.name == cond.name, cond.name is None),
    )

def awesome_build_query_with_dynamic_filter(session, cond: UserFilter):
    """
    a helper func, 根据 cond 返回一个 orm Query with dynamic filter

    这种写法利用getattr实现“自动把非None字段作为过滤条件”，当增减查询条件时，内部无需变化
    """
    
    # 生成字典并去掉None字段
    clean_filter = cond.dict(exclude_none=True)
    # 构建orm Query
    q = session.query(User)
    for attr, value in clean_filter.items():
        q = q.filter(getattr(User, attr)==value)
    return q

---

Because sqlalchemy uses magic methods (operator overloading) to create SQL constructs, it can only handle operator such as != or == in filter, and is not able to work with is (which is a very valid Python construct).

Therefore, to make it work with sqlalchemy, you should use:

...filter(or_(people.marriage_status!='married', people.marriage_status == None))

or 

...filter(or_(people.marriage_status!='married', people.marriage_status.is_(None)))

, basically replace the is None with == None or is_(None). In this case your query will be translated properly to the following SQL:

SELECT people.name AS people_name, people.marriage_status AS people_marriage_status 
FROM people 
WHERE people.marriage_status IS NULL OR people.marriage_status != ?

flask-sqlalchemy

Flask-SQLAlchemy provides a session object that is scoped to the current Flask application context. This means that each request gets its own session object that is created when the request starts and destroyed when the request ends. The session object is used to manage transactions with the database.

When you create a Flask-SQLAlchemy application named db, you can use the db.session object to interact with the database. For example, you can use it to query the database, add new rows, update rows, delete rows, etc.

flask-sqlalchemy封装的session默认是scoped_session，即thread local storage(TLS)

basic usage:

from flask_sqlalchemy import SQLAlchemy

# get a SQLAlchemy instance as db
db = SQLAlchemy(app)

# db.session is :class:`_orm.Session` object, which features **autobegin** behavior
# flask-sqlalchemy 中的 db.session 默认自动开启一个事务。您必须提交会话，但是没有必要在每个请求后删除它(session)，Flask-SQLAlchemy 会帮您完成删除操作
# 也就是说，无需显式开启事务，也无需显式关闭session。。。嗯，坑
# 什么时候会隐式开启事务呢？
# a transaction that is begun automatically when a statement is first invoked
# 所以线程内第一次直接查询，修改和删除都隐式地起了一个事务，坑……

# insert
db.session.add(admin) # 隐式开启了一个事务。。。该事务没commit前又执行db.session.begin()则会报错
db.session.add(guest)
db.session.commit()

or 

with db.session.begin(): # this autocommits if success, else auto rollback.
    db.session.add(admin)
    db.session.add(guest)

# query
with db.session.begin():
    res = db.session.query(User).filter(User.id == 1).first() # 查一条
    # res = db.session.query(User).filter(User.id == 1).all() # 查多条

# update
with db.session.begin():
     db.session.query(User).filter(User.id == 1).update({"age": 99})

flask-sqlalchemy中的隐式操作都应该抛弃

Always explicit begin a transaction and commit, No implicit begin transaction and commit
Always one choice - work with explicit connection, explicit transaction

the difference between the usage of methods query.with_entities and db.session.query in Flask-SQLAlchemy

In Flask-SQLAlchemy, both with_entities and db.session.query are used to define query objects and specify the entities to select from the database, but they are used in different stages of constructing the query.

db.session.query is the starting point of constructing a query, where you provide the SQLAlchemy models or columns you want to select as arguments. It creates a base query object with the entities you provide.

with_entities, on the other hand, is used on an existing query object to refine the query, replacing the previously set entities with the ones you provide as arguments.
也就是说，with_entities会完全覆盖db.session.query指定的查询字段

Here's an example to illustrate the difference:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'
db = SQLAlchemy(app)

# Assuming you already have your Employee class defined with id, name, and age fields

# Example 1: Using db.session.query
query1 = db.session.query(Employee.name)  # selects only the 'name' column from the Employee table

# Example 2: Using with_entities on an existing query
query2 = db.session.query(Employee) # selects all columns from the Employee table by default
# query2 = Employee.query  # the same as above
query2 = query2.with_entities(Employee.name)  # refines the query to select only the 'name' column

# Execute the queries
result1 = query1.all()
result2 = query2.all()

# Print results
print("Result 1: Using db.session.query")
for row in result1:
    print(row.name)

print("\nResult 2: Using with_entities")
for row in result2:
    print(row.name)

The main difference in the usage of with_entities and db.session.query in Flask-SQLAlchemy lies in their roles in constructing a query:

• db.session.query: The starting point to create a query with the specified entities.
• with_entities: Refined the entities to select on an existing query object.

注意，如果通过db.session.query或者with_entities指定了自定义的若干具体返回字段，那么db返回的结果将被封装成list[<class 'sqlalchemy.engine.row.Row'>]

the type of flask-sqlalchemy returned value after a query using join

When using Flask-SQLAlchemy, the returned value after executing a query with a join is a list of named tuples, where each named tuple represents a record fetched from the database.

Each named tuple contains the instances of the SQLAlchemy models that were fetched using the query.

Here's an example using Flask-SQLAlchemy:

# Import necessary modules/packages
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

# Assuming you have your Employee and Department classes defined using Flask-SQLAlchemy BaseModel
# ...

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///db.sqlite'
db = SQLAlchemy(app)

# ... Table creation (Employee and Department) with Employee related to Department ...

# Create a basic query
query = db.session.query(Employee, Department)
# Create your query using join
query = query.join(Department, Employee.department_id == Department.id)
results = query.all()

# Show the results
for row in results: # row is an instance of <class 'sqlalchemy.engine.row.Row'>, which is a named tuple
    employee, department = row # 用法1: 解构named tuple
    print(employee.name, department.name)

    print(row.Employee.name, row.Department.name) # 用法2: 直接通过name访问named tuple元素

results is a list of named tuples containing instances of Employee and Department