当前位置: 代码迷 >> python >> 如何为 DateTime 比较定义 Flask-SQLAlchemy(MySQL) 混合属性或聚合属性
  详细解决方案

如何为 DateTime 比较定义 Flask-SQLAlchemy(MySQL) 混合属性或聚合属性

热度:92   发布时间:2023-07-16 10:52:12.0

例如,数据库中有任务(父)对象和研究对象(子)。 研究对象具有开始和结束时间属性。 如何为 Task 对象定义两个混合属性。 一个是其任务中最早的开始时间,另一个是其任务中的最晚完成时间。

在文档中,它说聚合属性或混合属性可用于定义那些最早/最新的属性,但我无法管理如何比较 sa.DateTime 对象以及如何为这些对象编写 sql 查询。

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.sql import func as safunc

sa=SQLAlchemy()


class Task(sa.Model):
    __tablename__ = "tasks"
    id = sa.Column(sa.Integer, primary_key=True)
    studies = sa.relationship("Study", backref="task", cascade="all, delete, delete-orphan")

    @hybrid_propery
    def start(self):
        # ???

    @start.expression
    def start(cls):
        # ???

    @hybrid_propery
    def last_progress(self):
        # ???

    @last_progress.expression
    def last_progress(cls):
        # ???


class Study(sa.Model):
    __tablename__ = "studies"
    id = sa.Column(sa.Integer, primary_key=True)
    task_id = sa.Column(sa.Integer, sa.ForeignKey("tasks.id"))
    start = sa.Column(sa.DateTime, server_default=safunc.now())
    finish = sa.Column(sa.DateTime, nullable=True)

我处理它使用混合属性中的 python min/max 方法和属性表达式中 sqlalchemy.sql.func 中的 min/max funcs,代码工作正常。

task = Task.query.filter_by(id=id).first()

task.start查询返回最早的学习开始记录, task.last_progress返回我所寻找的该任务的最新学习完成时间。

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.sql import func as safunc

sa=SQLAlchemy()


class Task(sa.Model):
    __tablename__ = "tasks"
    id = sa.Column(sa.Integer, primary_key=True)
    studies = sa.relationship("Study", backref="task", cascade="all, delete, delete-orphan")

    @hybrid_propery
    def start(self):
        if any(i.start for i in self.studies):
            return min(filter(lambda x: x, [i.start for i in self.studies]))
        else:
            return None

    @start.expression
    def start(cls):
        return select([safunc.min(Study.start)]).\
                where(Study.task_id==cls.id).\
                label("start")

    @hybrid_propery
    def last_progress(self):
        if any(i.finish for i in self.studies):
            return max(filter(lambda x: x, [i.finish for i in self.studies]))
        else:
            return None

    @last_progress.expression
    def last_progress(cls):
        return select([safunc.max(Study.finish)]).\
                where(Study.task_id==cls.id).\
                label("last_progress")


class Study(sa.Model):
    __tablename__ = "studies"
    id = sa.Column(sa.Integer, primary_key=True)
    task_id = sa.Column(sa.Integer, sa.ForeignKey("tasks.id"))
    start = sa.Column(sa.DateTime, server_default=safunc.now())
    finish = sa.Column(sa.DateTime, nullable=True)
  相关解决方案