当前位置：首页 > news >正文

SQLAlchemy 中的 func 函数使用指南

news 2025/7/4 18:20:40

func 是 SQLAlchemy 中用于调用 SQL 函数的关键工具，它允许你在查询中使用数据库的各种内置函数。

基本介绍

func 是 sqlalchemy.sql.functions 模块中的一个特殊对象，它作为 SQL 函数的命名空间使用。当你访问 func 的属性时，SQLAlchemy 会创建一个对应的 SQL 函数调用。

常用函数示例

1. 聚合函数

from sqlalchemy import func
from models import User, Sessionsession = Session()# 计数
total_users = session.query(func.count(User.id)).scalar()# 求和
total_salary = session.query(func.sum(User.salary)).filter(User.department == 'IT').scalar()# 平均值
avg_age = session.query(func.avg(User.age)).scalar()# 最大值/最小值
max_salary = session.query(func.max(User.salary)).scalar()
min_age = session.query(func.min(User.age)).scalar()

2. 字符串函数

# 字符串连接
session.query(func.concat(User.first_name, ' ', User.last_name).label('full_name')).all()# 小写转换
session.query(func.lower(User.username)).all()# 子字符串
session.query(func.substr(User.email, 1, 5)).all()# 字符串长度
session.query(func.length(User.address)).all()

3. 日期时间函数

from datetime import datetime# 当前日期时间
session.query(func.now()).scalar()# 提取年份
session.query(func.extract('year', User.created_at)).all()# 日期格式化
session.query(func.strftime('%Y-%m-%d', User.birth_date)).all()# 日期差
session.query(func.datediff('day', User.start_date, User.end_date)).all()

4. 数学函数

# 随机数
session.query(User).order_by(func.random()).first()  # PostgreSQL/SQLite
session.query(User).order_by(func.rand()).first()    # MySQL# 绝对值
session.query(func.abs(User.balance)).all()# 四舍五入
session.query(func.round(User.score, 2)).all()# 平方根
session.query(func.sqrt(User.area)).all()

5. 条件表达式

# CASE WHEN 表达式
session.query(func.case([(User.age < 18, 'minor'),(User.age >= 18, 'adult')],else_='unknown').label('age_group')
).all()# COALESCE (返回第一个非NULL值)
session.query(func.coalesce(User.nickname, User.first_name)).all()# NULLIF (两值相等返回NULL)
session.query(func.nullif(User.salary, 0)).all()

高级用法

1. 函数组合

# 计算每个用户邮箱域名的长度
session.query(func.length(func.substr(User.email,func.instr(User.email, '@') + 1))
).all()

2. 自定义函数标签

# 为函数结果指定列名
session.query(func.upper(User.last_name).label('last_name_upper'),func.lower(User.first_name).label('first_name_lower')
).all()

3. 分组中使用函数

# 按年份统计用户数量
session.query(func.extract('year', User.created_at).label('year'),func.count(User.id).label('user_count')
).group_by('year').all()

4. 窗口函数

from sqlalchemy import over# 计算每个部门的薪资排名
session.query(User.name,User.salary,func.rank().over(order_by=User.salary.desc(),partition_by=User.department).label('rank')
).all()

数据库特定函数

SQLAlchemy 可以透明地处理不同数据库的函数差异：

# 处理不同数据库的日期差异计算
if dialect.name == 'postgresql':age_expr = func.age(User.end_date, User.start_date)
elif dialect.name == 'mysql':age_expr = func.timestampdiff('day', User.start_date, User.end_date)
else:age_expr = func.julianday(User.end_date) - func.julianday(User.start_date)

自定义函数

你还可以注册和使用自定义SQL函数：

from sqlalchemy import create_engine
from sqlalchemy.sql.expression import FunctionElement
from sqlalchemy.ext.compiler import compiles# 定义自定义函数类
class my_random(FunctionElement):type = Numeric()name = 'my_random'# 定义编译规则
@compiles(my_random, 'postgresql')
def compile_my_random_postgresql(element, compiler, **kw):return "my_random()"  # 调用PostgreSQL中的自定义函数# 使用自定义函数
session.query(User).order_by(my_random()).limit(5).all()