from tortoise import fields, models from tortoise.indexes import Index from tortoise.contrib.pydantic import pydantic_model_creator from tortoise.signals import post_save from typing import Type from helper.tools import genRandomStr # from conf.setting import setting class Agv(models.Model): id = fields.UUIDField(pk=True) num = fields.CharField(null=True, max_length=255, description='短编号') agvid = fields.CharField(null=True, max_length=255, description='AGV小车编号') percent = fields.DecimalField(default=1.0, max_digits=3, decimal_places=2, description='小车电量') records = fields.JSONField(null=True, description='记录列表') curorder = fields.CharField(null=True, max_length=255, description='当前订单号') curstop = fields.CharField(null=True, max_length=255, description='当前站点') curitinerary = fields.JSONField(null=True, description='订单对应的全局最短路径') is_charging = fields.BooleanField(default=False, description='充电中') is_busy = fields.BooleanField(default=False, description='设备忙') is_abnormal = fields.BooleanField(default=False, description='是否异常') abnormal_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 Meta: table = 'Agvs' table_description = 'AGV管理表' indexes = [ Index(fields={"num"}, name="num"), Index(fields={"agvid"}, name="agvid"), Index(fields={"is_busy"}, name="agvs_is_busy"), Index(fields={"is_abnormal"}, name="agvs_is_abnormal"), Index(fields={"is_done"}, name="agvs_is_done"), Index(fields={"is_charging"}, name="agvs_is_charging"), Index(fields={"is_valid"}, name="is_valididx"), Index(fields={"is_active"}, name="is_activeidx"), ] def __str__(self): return f'Agv(id={self.id} num={self.num}, agvid={self.agvid}, created_at={self.created_at})' def as_json(self): return dict(id="{}".format(self.id), num=self.num, agvid=self.agvid, created_at=self.created_at) Agv_Pydantic = pydantic_model_creator(Agv, name="Agv") @post_save(Agv) async def agv_post_save( sender: "Type[Agv]", instance: Agv, 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 Agv.filter(num = num).exists(): # print("num exist, ", num) checkNum() else: # print("num not exist, ", num) instance.num = num await instance.save() await checkNum()