Codex
95 lines
2.8 KiB
Python
95 lines
2.8 KiB
Python
#!/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
|