hd-crane-gs/backend/app/api/v1/module_business/gateway/schema.py

# -*- coding: utf-8 -*-

from pydantic import BaseModel, ConfigDict, Field
from fastapi import Query

from app.core.validator import DateTimeStr
from app.core.base_schema import BaseSchema, UserBySchema

class GatewayCreateSchema(BaseModel):
    """
    网关信息新增模型
    """
    crane_no: str = Field(default=..., description='行车编号')
    gateway_name: str = Field(default=..., description='网关名称')
    gateway_type: int = Field(default=..., description='网关类型')
    gateway_ipaddress: str = Field(default=..., description='网关IP地址 ')
    gateway_port: int | None = Field(default=None, description='网关端口 ')
    plc_brand: int | None = Field(default=None, description='PLC品牌')
    plc_model: str| None = Field(default=None, description='PLC型号 ')
    serial_port_name: str | None = Field(default=None, description='端口号')
    serial_baud_rate: int | None = Field(default=None, description='波特率 ')
    serial_data_bits: int | None = Field(default=None, description='数据位 5678')
    serial_stop_bits: int | None = Field(default=None, description='停止位 ')
    serial_parity: int | None = Field(default=None, description='检验位 ')
    status: str = Field(default="1", description='是否启用')
    description: str | None = Field(default=None, max_length=255, description='备注/描述')


class GatewayUpdateSchema(GatewayCreateSchema):
    """
    网关信息更新模型
    """
    ...


class GatewayOutSchema(GatewayCreateSchema, BaseSchema, UserBySchema):
    """
    网关信息响应模型
    """
    model_config = ConfigDict(from_attributes=True)


class GatewayQueryParam:
    """网关信息查询参数"""

    def __init__(
        self,
        gateway_name: str | None = Query(None, description="网关名称"),
        gateway_ipaddress: str | None = Query(None, description="网关IP地址 "),
        plc_model: str | None = Query(None, description="PLC型号 "),
        serial_port_name: str | None = Query(None, description="端口号"),
        status: str | None = Query(None, description="是否启用(0:启用 1:禁用)"),
        crane_no: str | None = Query(None, description="行车编号"),
        gateway_type: int | None = Query(None, description="网关类型"),
        gateway_port: int | None = Query(None, description="网关端口 "),
        plc_brand: int | None = Query(None, description="PLC品牌"),
        serial_baud_rate: int | None = Query(None, description="波特率 "),
        serial_data_bits: int | None = Query(None, description="数据位 5678"),
        serial_stop_bits: int | None = Query(None, description="停止位 "),
        serial_parity: int | None = Query(None, description="检验位 "),
        created_time: list[DateTimeStr] | None = Query(None, description="创建时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"]),
        updated_time: list[DateTimeStr] | None = Query(None, description="更新时间范围", examples=["2025-01-01 00:00:00", "2025-12-31 23:59:59"]),
        
    ) -> None:
        
        # 精确查询字段
        self.crane_no = crane_no
        # 模糊查询字段
        self.gateway_name = ("like", gateway_name)
        # 精确查询字段
        self.gateway_type = gateway_type
        # 模糊查询字段
        self.gateway_ipaddress = ("like", gateway_ipaddress)
        # 精确查询字段
        self.gateway_port = gateway_port
        # 精确查询字段
        self.plc_brand = plc_brand
        # 模糊查询字段
        self.plc_model = ("like", plc_model)
        # 模糊查询字段
        self.serial_port_name = ("like", serial_port_name)
        # 精确查询字段
        self.serial_baud_rate = serial_baud_rate
        # 精确查询字段
        self.serial_data_bits = serial_data_bits
        # 精确查询字段
        self.serial_stop_bits = serial_stop_bits
        # 精确查询字段
        self.serial_parity = serial_parity
        # 模糊查询字段
        self.status = ("like", status)
        # 时间范围查询
        if created_time and len(created_time) == 2:
            self.created_time = ("between", (created_time[0], created_time[1]))
        if updated_time and len(updated_time) == 2:
            self.updated_time = ("between", (updated_time[0], updated_time[1]))