26 lines
1.2 KiB
Python
26 lines
1.2 KiB
Python
from typing import Any, Dict, Optional
|
||
from pydantic import BaseModel, Field
|
||
|
||
class ConnectionPayload(BaseModel):
|
||
server: str = Field(..., example="mssql.example.local,1433")
|
||
database: str = Field(..., example="MyDb")
|
||
username: Optional[str] = Field(None, example="sa")
|
||
password: Optional[str] = Field(None, example="S3cret!")
|
||
# For servers with encryption disabled, set encrypt=False (default below)
|
||
encrypt: bool = Field(default=False, description="ODBC Encrypt setting: yes/no")
|
||
trust_server_certificate: bool = Field(
|
||
default=True,
|
||
description="If encrypt=True and you don’t have CA chain, set True"
|
||
)
|
||
timeout: int = Field(default=15, ge=1, le=120)
|
||
|
||
class QueryPayload(ConnectionPayload):
|
||
sql: str = Field(..., example="SELECT TOP 5 id, name FROM dbo.Customers WHERE city = :city")
|
||
params: Dict[str, Any] = Field(default_factory=dict)
|
||
max_rows: Optional[int] = Field(None, ge=1, example=1000)
|
||
|
||
class ExecutePayload(ConnectionPayload):
|
||
sql: str = Field(..., example="UPDATE dbo.Customers SET name = :name WHERE id = :id")
|
||
params: Dict[str, Any] = Field(default_factory=dict)
|
||
autocommit: bool = Field(default=False)
|