#!/usr/bin/env python # -*- coding: utf-8 -*- """数据库连接 Schema""" from typing import Any, Dict, Optional from pydantic import BaseModel, Field from app.base_schema import CSTDatetime class DatabaseConnectionBase(BaseModel): name: str = Field(..., max_length=100, description="连接名称") code: str = Field(..., max_length=50, description="连接编码") db_type: str = Field(..., description="postgresql / mysql / sqlserver / oracle") host: str = Field(..., max_length=255) port: int = Field(default=5432, ge=1, le=65535) user: str = Field(default="", max_length=100) default_database: str = Field(default="", max_length=100, alias="defaultDatabase") description: str = Field(default="") status: bool = Field(default=True) application_id: Optional[str] = Field(default=None, alias="applicationId") extra_options: Optional[Dict[str, Any]] = Field(default_factory=dict, alias="extraOptions") model_config = {"populate_by_name": True} class DatabaseConnectionCreate(DatabaseConnectionBase): password: str = Field(default="", description="密码(明文,仅创建时传入)") class DatabaseConnectionUpdate(BaseModel): name: Optional[str] = None db_type: Optional[str] = None host: Optional[str] = None port: Optional[int] = Field(default=None, ge=1, le=65535) user: Optional[str] = None password: Optional[str] = Field(default=None, description="空则不修改") default_database: Optional[str] = Field(default=None, alias="defaultDatabase") description: Optional[str] = None status: Optional[bool] = None extra_options: Optional[Dict[str, Any]] = Field(default=None, alias="extraOptions") model_config = {"populate_by_name": True} class DatabaseConnectionResponse(BaseModel): id: str code: str name: str db_type: str host: str port: int user: str default_database: str description: str status: bool is_system: bool has_password: bool application_id: Optional[str] = None application_name: str = "" extra_options: Dict[str, Any] = Field(default_factory=dict) sort: int = 0 sys_create_datetime: Optional[CSTDatetime] = None sys_update_datetime: Optional[CSTDatetime] = None model_config = {"from_attributes": True} class DatabaseConnectionSimpleOut(BaseModel): code: str name: str db_type: str host: str port: int database: str is_system: bool = False class DatabaseConnectionTestRequest(BaseModel): db_type: str host: str port: int = 5432 user: str = "" password: str = "" default_database: str = Field(default="", alias="defaultDatabase") model_config = {"populate_by_name": True} class DatabaseConnectionTestResponse(BaseModel): success: bool message: str db_name: Optional[str] = None db_type: Optional[str] = None