chore(api): convert RecommendAppRetrievalBase and WorkflowPauseEntity from ABC to Protocol (#37182)

This commit is contained in:
Eric Cao 2026-06-08 22:17:07 +08:00 committed by GitHub
parent 0239b81cca
commit 5d814ca8c1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 12 additions and 23 deletions

View File

@ -6,16 +6,16 @@ by the core workflow module. These models are independent of the storage mechani
and don't contain implementation details like tenant_id, app_id, etc.
"""
from abc import ABC, abstractmethod
from collections.abc import Sequence
from datetime import datetime
from typing import Protocol
from graphon.entities.pause_reason import PauseReason
class WorkflowPauseEntity(ABC):
class WorkflowPauseEntity(Protocol):
"""
Abstract base class for workflow pause entities.
Protocol for workflow pause entities.
This domain model represents a paused workflow execution state,
without implementation details like tenant_id, app_id, etc.
@ -27,19 +27,17 @@ class WorkflowPauseEntity(ABC):
"""
@property
@abstractmethod
def id(self) -> str:
"""The identifier of current WorkflowPauseEntity"""
pass
...
@property
@abstractmethod
def workflow_execution_id(self) -> str:
"""The identifier of the workflow execution record the pause associated with.
Correspond to `WorkflowExecution.id`.
"""
...
@abstractmethod
def get_state(self) -> bytes:
"""
Retrieve the serialized workflow state from storage.
@ -56,20 +54,17 @@ class WorkflowPauseEntity(ABC):
...
@property
@abstractmethod
def resumed_at(self) -> datetime | None:
"""`resumed_at` return the resumption time of the current pause, or `None` if
the pause is not resumed yet.
"""
pass
...
@property
@abstractmethod
def paused_at(self) -> datetime:
"""`paused_at` returns the creation time of the pause."""
pass
...
@abstractmethod
def get_pause_reasons(self) -> Sequence[PauseReason]:
"""
Retrieve detailed reasons for this pause.

View File

@ -1,17 +1,11 @@
from abc import ABC, abstractmethod
from typing import Any, Protocol
class RecommendAppRetrievalBase(ABC):
class RecommendAppRetrievalBase(Protocol):
"""Interface for recommend app retrieval."""
@abstractmethod
def get_recommended_apps_and_categories(self, language: str):
raise NotImplementedError
def get_recommended_apps_and_categories(self, language: str) -> Any: ...
@abstractmethod
def get_recommend_app_detail(self, app_id: str):
raise NotImplementedError
def get_recommend_app_detail(self, app_id: str) -> Any: ...
@abstractmethod
def get_type(self) -> str:
raise NotImplementedError
def get_type(self) -> str: ...