Developer
95 lines
2.4 KiB
Python
95 lines
2.4 KiB
Python
"""Pydantic models for the service registry API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
def _validate_config_keys(config: dict[str, Any]) -> dict[str, Any]:
|
|
"""Reject credential keys in non-secret service config.
|
|
|
|
Secrets are sent in the separate ``secrets`` mapping; the plain ``config``
|
|
object must never hold them.
|
|
"""
|
|
forbidden = {
|
|
"password",
|
|
"token",
|
|
"secret",
|
|
"api_key",
|
|
"apikey",
|
|
"private_key",
|
|
"passphrase",
|
|
"credential",
|
|
}
|
|
|
|
def _check(value: Any) -> None:
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
if key.lower() in forbidden:
|
|
raise ValueError(f"Credential key '{key}' is not allowed in service config")
|
|
_check(child)
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
_check(item)
|
|
|
|
_check(config)
|
|
return config
|
|
|
|
|
|
class ServiceInstanceInput(BaseModel):
|
|
"""Payload for creating or updating a service instance."""
|
|
|
|
id: str | None = None
|
|
service_type: str = Field(..., min_length=1)
|
|
name: str = Field(..., min_length=1)
|
|
config: dict[str, Any] = Field(default_factory=dict)
|
|
secrets: dict[str, str] = Field(default_factory=dict)
|
|
enabled: bool = True
|
|
|
|
@field_validator("config")
|
|
@classmethod
|
|
def reject_credential_keys(cls, value: dict[str, Any]) -> dict[str, Any]:
|
|
return _validate_config_keys(value or {})
|
|
|
|
|
|
class ServiceInstance(BaseModel):
|
|
"""Persisted service instance returned by the API (no plaintext secrets)."""
|
|
|
|
id: str
|
|
service_type: str
|
|
name: str
|
|
config: dict[str, Any]
|
|
secrets_set: dict[str, bool]
|
|
enabled: bool
|
|
created_at: int
|
|
updated_at: int
|
|
|
|
|
|
class SecretFieldInfo(BaseModel):
|
|
key: str
|
|
label: str
|
|
required: bool = False
|
|
helper: str | None = None
|
|
|
|
|
|
class WidgetKindInfo(BaseModel):
|
|
kind: str
|
|
name: str
|
|
description: str
|
|
config_schema: dict[str, Any]
|
|
default_config: dict[str, Any]
|
|
refresh_interval_ms: int
|
|
|
|
|
|
class ServiceTypeInfo(BaseModel):
|
|
"""Metadata about a registered service type."""
|
|
|
|
service_type: str
|
|
name: str
|
|
description: str
|
|
config_schema: dict[str, Any]
|
|
secret_fields: list[SecretFieldInfo]
|
|
widget_kinds: list[WidgetKindInfo]
|