from tortoise import fields, models from tortoise.indexes import Index from tortoise.contrib.pydantic import pydantic_model_creator, pydantic_queryset_creator from tortoise.signals import post_save from typing import Type from helper.tools import genRandomStr # from conf.setting import setting # 最小可执行子任务 class Subtask(models.Model): id = fields.UUIDField(pk=True) num = fields.CharField(null=True, max_length=255, description='短编号') orderid = fields.CharField(null=True, max_length=255, description='中控系统订单编号') owner = fields.ForeignKeyField('models.User', on_delete=fields.CASCADE, null=True, description='订单所有人') sequence = fields.IntField(default=0, description='管理员手动排序号') #取预约领用的动作类型属于现场领用 action = fields.IntField(default=0, description='1: 归还, 3: 预约领用, 5: 现场领用') ordertime = fields.DatetimeField(null=True, description='预约领用时间') name = fields.CharField(null=True, max_length=255, description='试剂名称') #取预约领用的任务类型属于非标标准品 typetask = fields.IntField(default=1, description='1: 标准品A, 3: 标准品B, 5: 标准品C, 7: 标准品D, 9:非标准品') quantity = fields.IntField(default=0, description='试剂数量') #坐标格式： 停靠点 - 层坐标 (最底下为第一层) - 试剂在托盘内的编号 coordinates = fields.CharField(null=True, max_length=255, description='试剂坐标') traynum = fields.CharField(null=True, max_length=255, description='托盘编号') is_canceled = fields.BooleanField(default=False, description='是否取消') cancel_time = fields.DatetimeField(null=True, description='取消时间') is_done = fields.BooleanField(default=False, description='是否完成') created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) is_valid = fields.BooleanField(default=True, description='是否有效') is_active = fields.BooleanField(default=True, description='是否删除') class PydanticMeta: # Let's exclude the created timestamp exclude = ("created_at",) class Meta: table = 'Subtasks' table_description = '子任务表' #set the ordering where sequence > action > ordertime > typetask #when passing to order_by function ordering = ["-sequence", "-action", "ordertime"] indexes = [ Index(fields={"num"}, name="numidx"), Index(fields={"orderid"}, name="orderididx"), Index(fields={"ordertime"}, name="subtask_ordertimeidx"), Index(fields={"is_canceled"}, name="subtask_is_canceledidx"), Index(fields={"is_done"}, name="subtask_is_doneidx"), Index(fields={"coordinates"}, name="subtask_coordinatesidx"), Index(fields={"traynum"}, name="subtask_traynumidx"), Index(fields={"is_valid"}, name="is_valididx"), Index(fields={"is_active"}, name="is_activeidx"), ] def __str__(self): return f'Subtask(id={self.id} num={self.num}, name={self.name}, quantity={self.quantity}, action={self.action}, typetask={self.typetask}, created_at={self.created_at})' def as_json(self): return dict(id="{}".format(self.id), num=self.num, name=self.name, quantity=self.quantity, action=self.action, typetask=self.typetask, created_at="{}".format(self.created_at)) Subtask_Pydantic = pydantic_model_creator(Subtask, name="Subtask") Subtask_Pydantic_List = pydantic_queryset_creator(Subtask) @post_save(Subtask) async def subtask_post_save( sender: "Type[Subtask]", instance: Subtask, created, using_db, update_fields, ) -> None: if not instance.num or "null" == instance.num: # print("instance.num: ", instance.num) # check duplicated num async def checkNum(): num = genRandomStr(4) # print("new num: ", num) if await Subtask.filter(num = num).exists(): # print("num exist, ", num) checkNum() else: # print("num not exist, ", num) instance.num = num await instance.save() await checkNum() # 执行任务 class Task(models.Model): id = fields.UUIDField(pk=True) num = fields.CharField(null=True, max_length=255, description='短编号') orderid = fields.CharField(null=True, max_length=255, description='中控系统订单编号') owner = fields.ForeignKeyField('models.User', on_delete=fields.CASCADE, null=True, description='订单所有人') sequence = fields.IntField(default=0, description='管理员手动排序号') #取预约领用的动作类型属于现场领用 action = fields.IntField(default=0, description='1: 归还, 3: 预约领用, 5: 现场领用') ordertime = fields.DatetimeField(null=True, description='预约领用时间') name = fields.CharField(null=True, max_length=255, description='试剂名称') #取预约领用的任务类型属于非标标准品 typetask = fields.IntField(default=1, description='1: 标准品A, 3: 标准品B, 5: 标准品C, 7: 标准品D, 9:非准品') quantity = fields.IntField(default=0, description='试剂数量') #坐标格式： 停靠点 - 层坐标 (最底下为第一层) - 试剂在托盘内的编号 coordinates = fields.CharField(null=True, max_length=255, description='试剂坐标') traynum = fields.CharField(null=True, max_length=255, description='托盘编号') subtasks = fields.ManyToManyField('models.Subtask', related_name='tasksubtasks') is_canceled = fields.BooleanField(default=False, description='是否取消') cancel_time = fields.DatetimeField(null=True, description='取消时间') is_done = fields.BooleanField(default=False, description='是否完成') created_at = fields.DatetimeField(auto_now_add=True) updated_at = fields.DatetimeField(auto_now=True) is_valid = fields.BooleanField(default=True, description='是否有效') is_active = fields.BooleanField(default=True, description='是否删除') class PydanticMeta: # Let's exclude the created timestamp exclude = ("created_at",) allow_cycles = True max_recursion = 4 class Meta: table = 'Tasks' table_description = '任务表' #set the ordering where sequence > action > ordertime > typetask #when passing to order_by function ordering = ["-sequence", "-action", "ordertime"] indexes = [ Index(fields={"num"}, name="numidx"), Index(fields={"orderid"}, name="orderididx"), Index(fields={"ordertime"}, name="task_ordertimeidx"), Index(fields={"is_canceled"}, name="task_is_canceledidx"), Index(fields={"is_done"}, name="task_is_doneidx"), Index(fields={"coordinates"}, name="task_coordinatesidx"), Index(fields={"traynum"}, name="task_traynumidx"), Index(fields={"is_valid"}, name="is_valididx"), Index(fields={"is_active"}, name="is_activeidx"), ] def __str__(self): return f'Task(id={self.id} num={self.num}, name={self.name}, quantity={self.quantity}, action={self.action}, typetask={self.typetask}, created_at={self.created_at})' def as_json(self): return dict(id="{}".format(self.id), num=self.num, name=self.name, quantity=self.quantity, action=self.action, typetask=self.typetask, created_at="{}".format(self.created_at)) Task_Pydantic = pydantic_model_creator(Task, name="Task") Task_Pydantic_List = pydantic_queryset_creator(Task) @post_save(Task) async def task_post_save( sender: "Type[Task]", instance: Task, created, using_db, update_fields, ) -> None: if not instance.num or "null" == instance.num: # print("instance.num: ", instance.num) # check duplicated num async def checkNum(): num = genRandomStr(4) # print("new num: ", num) if await Task.filter(num = num).exists(): # print("num exist, ", num) checkNum() else: # print("num not exist, ", num) instance.num = num await instance.save() await checkNum()