“

系列：S6 平台工程 · 第 13 篇 | 难度：高级 | 阅读时间：18 分钟

SDK 设计哲学：Ontology-First 的开发者体验

#TL;DR

coomia-dip SDK 的设计哲学是"Ontology-First"——开发者通过 Ontology 对象模型而非底层 API 与平台交互。SDK 提供类型安全的代码生成、直觉式的流式 API、自动化的权限上下文传播和透明的 gRPC 通信。本文从设计原则、分层架构、代码生成策略、错误处理哲学到版本兼容性，完整阐述 coomia-dip SDK 的设计思想和工程决策。

#1. 设计原则

#1.1 五大设计原则

coomia-dip SDK 的设计遵循五大核心原则：

Code

┌─────────────────────────────────────────┐
│          SDK Design Principles           │
│                                          │
│  1. Ontology-First  (本体优先)            │
│  2. Type-Safe       (类型安全)            │
│  3. Zero-Boilerplate(零模板代码)          │
│  4. Fail-Fast       (快速失败)            │
│  5. Transparent     (透明通信)            │
└─────────────────────────────────────────┘

原则 1：Ontology-First（本体优先）

开发者操作的是 Ontology 对象（Employee、Project、Action），而非 REST 端点或 gRPC 方法。SDK 将平台能力封装在对象模型背后：

Python

# 反模式：底层 API 调用
response = client.post("/api/v1/objects/Employee/query", json={...})

# coomia-dip SDK：Ontology-First
employees = await client.Employee.where(
    Employee.department == "Engineering"
).select(
    Employee.name, Employee.salary
).limit(100).execute()

原则 2：Type-Safe（类型安全）

所有 Ontology 操作在编译时进行类型检查。属性名称、类型和约束通过代码生成嵌入 SDK：

Python

# 类型错误在 IDE 中立即显示
employee.salary = "not_a_number"  # TypeError: expected int, got str
employee.nonexistent_field = 42   # AttributeError: Employee has no property 'nonexistent_field'

原则 3：Zero-Boilerplate（零模板代码）

连接配置、认证、序列化、错误重试等基础设施代码由 SDK 自动处理：

Python

# 一行初始化
client = OntoPlatform.connect("https://platform.example.com", token="...")

# 无需手动构造请求/解析响应
employee = await client.Employee.get("emp-001")

原则 4：Fail-Fast（快速失败）

错误在最早的时间点被检测和报告，提供清晰的错误消息和修复建议：

Python

try:
    await client.Employee.create(name="John")  # 缺少必填字段 department
except ValidationError as e:
    # "Employee.create requires field 'department' (str).
    #  Provided fields: ['name']. Missing required: ['department', 'hire_date']"

原则 5：Transparent（透明通信）

SDK 内部的 gRPC 通信对开发者透明，但在需要时可以观察和调试：

Python

# 启用请求日志
client = OntoPlatform.connect(..., debug=True)
# DEBUG: gRPC call OntologyService.GetObject(type=Employee, id=emp-001) -> 200 (12ms)

#1.2 对标 Palantir OSDK

设计维度	Palantir OSDK	coomia-dip SDK
核心抽象	Ontology 对象	Ontology 对象
类型安全	TypeScript 生成	Python + TS 生成
通信协议	REST/JSON	gRPC/Protobuf
代码生成	CLI 工具	CLI + CI 集成
异步支持	Promise	async/await 原生

#2. 分层架构

#2.1 SDK 四层架构

Code

┌─────────────────────────────────────────┐
│  Layer 4: Generated Ontology Layer      │
│  (自动生成的类型安全对象模型)              │
│  Employee, Project, CreateEmployee...   │
├─────────────────────────────────────────┤
│  Layer 3: Domain API Layer              │
│  (领域操作 API)                          │
│  ObjectClient, ActionClient,            │
│  SearchClient, LinkClient               │
├─────────────────────────────────────────┤
│  Layer 2: Transport Layer               │
│  (gRPC 通信 + 序列化)                    │
│  GrpcChannel, Interceptors,             │
│  Serializer, RetryPolicy                │
├─────────────────────────────────────────┤
│  Layer 1: Foundation Layer              │
│  (基础设施)                              │
│  Auth, Config, Logging,                 │
│  Connection Pool, Circuit Breaker       │
└─────────────────────────────────────────┘

#2.2 Layer 1：基础层

Python

class SDKConfig(BaseModel):
    """SDK 配置"""

    platform_url: str = Field(description="平台地址")
    auth_token: str | None = Field(default=None)
    auth_provider: AuthProvider | None = Field(default=None)

    # 连接配置
    max_connections: int = Field(default=10)
    connect_timeout_ms: int = Field(default=5000)
    request_timeout_ms: int = Field(default=30000)

    # 重试配置
    max_retries: int = Field(default=3)
    retry_backoff_ms: int = Field(default=100)
    retry_max_backoff_ms: int = Field(default=5000)

    # 可观测性
    enable_tracing: bool = Field(default=False)
    enable_metrics: bool = Field(default=False)
    log_level: str = Field(default="INFO")


class ConnectionManager:
    """连接管理器 - 连接池 + 健康检查"""

    def __init__(self, config: SDKConfig):
        self._config = config
        self._channel: grpc.aio.Channel | None = None
        self._circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=30,
        )

    async def get_channel(self) -> grpc.aio.Channel:
        if self._channel is None:
            self._channel = grpc.aio.insecure_channel(
                self._config.platform_url,
                options=[
                    ("grpc.max_receive_message_length", 50 * 1024 * 1024),
                    ("grpc.keepalive_time_ms", 10000),
                ],
            )
        return self._channel

#2.3 Layer 2：传输层

Python

class AuthInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    """认证拦截器"""

    async def intercept_unary_unary(self, continuation, client_call_details, request):
        metadata = list(client_call_details.metadata or [])
        token = await self._auth_provider.get_token()
        metadata.append(("authorization", f"Bearer {token}"))

        new_details = client_call_details._replace(metadata=metadata)
        return await continuation(new_details, request)


class RetryInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    """重试拦截器"""

    RETRYABLE_CODES = {
        grpc.StatusCode.UNAVAILABLE,
        grpc.StatusCode.DEADLINE_EXCEEDED,
        grpc.StatusCode.RESOURCE_EXHAUSTED,
    }

    async def intercept_unary_unary(self, continuation, client_call_details, request):
        for attempt in range(self._max_retries + 1):
            try:
                return await continuation(client_call_details, request)
            except grpc.aio.AioRpcError as e:
                if e.code() not in self.RETRYABLE_CODES or attempt == self._max_retries:
                    raise
                await asyncio.sleep(self._backoff(attempt))


class TracingInterceptor(grpc.aio.UnaryUnaryClientInterceptor):
    """OpenTelemetry 追踪拦截器"""

    async def intercept_unary_unary(self, continuation, client_call_details, request):
        with tracer.start_as_current_span(
            f"grpc.{client_call_details.method}",
            kind=SpanKind.CLIENT,
        ) as span:
            span.set_attribute("rpc.system", "grpc")
            span.set_attribute("rpc.method", client_call_details.method)

            try:
                response = await continuation(client_call_details, request)
                span.set_status(StatusCode.OK)
                return response
            except grpc.aio.AioRpcError as e:
                span.set_status(StatusCode.ERROR, str(e))
                raise

#2.4 Layer 3：领域 API 层

Python

class ObjectClient:
    """对象操作客户端"""

    async def get(self, object_type: str, object_id: str) -> OntologyObject:
        ...

    async def list(
        self,
        object_type: str,
        filters: list[Filter] | None = None,
        order_by: list[OrderBy] | None = None,
        page_size: int = 100,
        page_token: str | None = None,
    ) -> PagedResult[OntologyObject]:
        ...

    async def create(self, object_type: str, properties: dict) -> OntologyObject:
        ...

    async def update(self, object_type: str, object_id: str, updates: dict) -> OntologyObject:
        ...

    async def delete(self, object_type: str, object_id: str) -> None:
        ...


class ActionClient:
    """Action 操作客户端"""

    async def execute(
        self,
        action_type: str,
        parameters: dict,
        mode: ExecutionMode = ExecutionMode.VALIDATE_AND_EXECUTE,
    ) -> ActionResult:
        ...

    async def validate(self, action_type: str, parameters: dict) -> ValidationResult:
        ...

#2.5 Layer 4：生成层

Python

# 自动生成的类型安全模型（示例）
class Employee(OntologyObject):
    """Employee 对象类型 - 自动生成"""

    __object_type__ = "Employee"

    # 类型安全的属性
    name: str
    department: str
    salary: int
    hire_date: date
    email: str | None = None
    manager_id: str | None = None

    # 类型安全的查询构建器
    class Properties:
        name = StringProperty("name")
        department = StringProperty("department")
        salary = IntProperty("salary")
        hire_date = DateProperty("hire_date")
        email = StringProperty("email")
        manager_id = StringProperty("manager_id")

    # 类型安全的关联
    async def manager(self) -> "Employee | None":
        if self.manager_id:
            return await self._client.Employee.get(self.manager_id)
        return None

    async def reports(self) -> list["Employee"]:
        return await self._client.Employee.where(
            Employee.Properties.manager_id == self.id
        ).execute()

#3. 代码生成策略

#3.1 生成流程

Code

Schema Registry → Protobuf Def → Code Generator → Type-Safe SDK
       │               │                │               │
  Ontology Schema   .proto 文件    Jinja2 模板       Python/TS 代码
  (运行时获取)     (中间表示)     (语言特定)        (开发者使用)

#3.2 生成器实现

Python

class SDKCodeGenerator:
    """SDK 代码生成器"""

    async def generate(
        self,
        schema_url: str,
        output_dir: str,
        language: str = "python",
    ) -> GenerationResult:
        """从 Schema Registry 生成类型安全的 SDK 代码"""
        # 1. 获取 Schema
        schemas = await self._schema_client.list_object_types()

        # 2. 生成代码
        generated_files = []
        for schema in schemas:
            template = self._get_template(language, schema.type)
            code = template.render(
                object_type=schema,
                properties=schema.properties,
                links=schema.links,
                actions=schema.actions,
            )
            file_path = f"{output_dir}/{schema.api_name.lower()}.py"
            generated_files.append(file_path)

        # 3. 生成索引文件
        index_code = self._generate_index(schemas, language)
        generated_files.append(f"{output_dir}/__init__.py")

        return GenerationResult(
            files=generated_files,
            object_types=len(schemas),
            language=language,
        )

#4. 错误处理哲学

#4.1 错误层级

Python

class OntoSDKError(Exception):
    """SDK 基础错误"""
    def __init__(self, message: str, code: str, details: dict | None = None):
        self.code = code
        self.details = details or {}
        super().__init__(message)

class ConnectionError(OntoSDKError):
    """连接错误"""
    pass

class AuthenticationError(OntoSDKError):
    """认证错误"""
    pass

class AuthorizationError(OntoSDKError):
    """授权错误 - 权限不足"""
    pass

class ValidationError(OntoSDKError):
    """验证错误 - 输入数据不合法"""
    def __init__(self, message: str, field_errors: dict[str, list[str]]):
        self.field_errors = field_errors
        super().__init__(message, code="VALIDATION_ERROR")

class ObjectNotFoundError(OntoSDKError):
    """对象未找到"""
    def __init__(self, object_type: str, object_id: str):
        super().__init__(
            f"{object_type} with id '{object_id}' not found",
            code="NOT_FOUND",
            details={"object_type": object_type, "object_id": object_id},
        )

class ConflictError(OntoSDKError):
    """冲突错误 - 并发修改"""
    pass

#4.2 错误映射

Python

class GrpcErrorMapper:
    """gRPC 错误码到 SDK 错误的映射"""

    MAPPING = {
        grpc.StatusCode.NOT_FOUND: ObjectNotFoundError,
        grpc.StatusCode.PERMISSION_DENIED: AuthorizationError,
        grpc.StatusCode.UNAUTHENTICATED: AuthenticationError,
        grpc.StatusCode.INVALID_ARGUMENT: ValidationError,
        grpc.StatusCode.ALREADY_EXISTS: ConflictError,
        grpc.StatusCode.UNAVAILABLE: ConnectionError,
    }

    def map(self, grpc_error: grpc.aio.AioRpcError) -> OntoSDKError:
        error_class = self.MAPPING.get(grpc_error.code(), OntoSDKError)
        return error_class(
            message=grpc_error.details(),
            code=grpc_error.code().name,
        )

#5. 版本兼容性策略

#5.1 语义版本控制

Code

SDK 版本：MAJOR.MINOR.PATCH
  MAJOR: 不兼容的 API 变更
  MINOR: 向后兼容的功能添加
  PATCH: 向后兼容的 bug 修复

Schema 版本：独立递增
  SDK 支持 Schema 版本范围（如 v5-v8）

#5.2 向后兼容保证

Python

class VersionNegotiator:
    """版本协商器"""

    async def negotiate(self, server_version: str) -> CompatibilityResult:
        sdk_version = self._get_sdk_version()

        if not self._is_compatible(sdk_version, server_version):
            return CompatibilityResult(
                compatible=False,
                message=f"SDK {sdk_version} is not compatible with server {server_version}. "
                        f"Please upgrade SDK to >= {self._min_required_sdk(server_version)}",
            )

        if self._has_deprecation_warnings(sdk_version, server_version):
            warnings = self._get_deprecation_warnings(sdk_version, server_version)
            return CompatibilityResult(
                compatible=True,
                warnings=warnings,
            )

        return CompatibilityResult(compatible=True)

#6. 测试策略

Python

class TestSDKDesign:
    async def test_ontology_first_api(self):
        """验证 Ontology-First API 风格"""
        client = OntoPlatform.connect(test_url, token=test_token)
        employee = await client.Employee.get("emp-001")
        assert isinstance(employee, Employee)
        assert hasattr(employee, "name")
        assert hasattr(employee, "salary")

    async def test_type_safety(self):
        """验证类型安全"""
        with pytest.raises(ValidationError):
            await client.Employee.create(
                name="John",
                salary="not_a_number",  # 应报类型错误
            )

    async def test_error_mapping(self):
        """验证错误映射"""
        with pytest.raises(ObjectNotFoundError) as exc_info:
            await client.Employee.get("nonexistent")
        assert "not found" in str(exc_info.value)

    async def test_retry_on_transient_error(self):
        """验证瞬态错误重试"""
        # 模拟前两次失败，第三次成功
        mock_channel = MockGrpcChannel(failures=2)
        client = OntoPlatform(channel=mock_channel)
        result = await client.Employee.get("emp-001")
        assert result is not None
        assert mock_channel.call_count == 3

#7. 总结

coomia-dip SDK 的设计哲学围绕五大原则构建，从底层基础设施到顶层类型安全模型，每一层都为开发者体验服务。关键设计决策：

Ontology-First：开发者操作对象模型，而非底层 API
类型安全：代码生成确保编译时类型检查
四层架构：Foundation → Transport → Domain → Generated
快速失败：清晰的错误层级和修复建议
版本兼容：语义版本 + 向后兼容保证

下一篇将深入探讨 gRPC 客户端代码生成的具体实现。