Skip to content

Commit

Permalink
🚧 refactored: remove Action and Function from legacy.
Browse files Browse the repository at this point in the history
  • Loading branch information
korawica committed May 10, 2024
1 parent 461ba32 commit 8bfd413
Show file tree
Hide file tree
Showing 9 changed files with 139 additions and 764 deletions.
26 changes: 13 additions & 13 deletions app/controls.py
Original file line number Diff line number Diff line change
Expand Up @@ -308,16 +308,16 @@ def push_testing() -> None:
logger.info("Start Testing ...")
with Task.make(module="demo_docstring"):
Node.parse_name(fullname="imp_min_max_service_level")
# (
# ActionQuery.parse_name(fullname="query:query_shutdown")
# .add_ext_params(
# params={
# "status": 1,
# "process_message": (
# "Error: RuntimeError: Server shutdown while "
# "process was running in background"
# ),
# },
# )
# .push()
# )
(
ActionQuery.parse_name(fullname="query:query_shutdown")
.add_ext_params(
params={
"status": 1,
"process_message": (
"Error: RuntimeError: Server shutdown while "
"process was running in background"
),
},
)
.execute()
)
127 changes: 0 additions & 127 deletions app/core/__legacy/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -1159,133 +1159,6 @@ def _generate_initial(self, initial: dict) -> dict:
return _initial


class FuncCatalog:
"""Function Catalog for generate all configuration from .yaml file."""

__slots__ = (
"func_name",
"func_type",
"func_name_sht",
"func_prefix",
"func_catalog",
)

def __init__(
self,
func_name: str,
func_type: Optional[str] = None,
func_catalog: Optional[dict] = None,
):
self.func_name: str = func_name
self.func_type: str = func_type or params.list_func_types[0]
self.func_name_sht: str = "".join(
[word[0] for word in self.func_name.split("_")]
)
self.func_prefix: str = self.func_name.split("_")[0]
self.func_catalog: dict = self.get_func_catalog(config=func_catalog)
if self.func_prefix not in params.list_func_priority:
logger.warning(
f"The config {self.func_name!r} have table prefix name, `{self.func_prefix}`, "
f"which different from the agreed list"
)

def __repr__(self):
return f"{self.__class__.__name__}({self.func_name!r}, func_type={self.func_type!r})"

def __str__(self):
return f"{self.func_type}:{self.func_name}"

# [x] Migrate to modern style
@property
def catalog(self) -> dict:
return {
"id": self.func_name_sht,
"name": self.func_name,
"type": self.func_type,
"prefix": self.func_prefix,
**self.func_catalog,
}

@property
def func_profile(self) -> dict:
return self.func_catalog["profile"]

def get_func_stm_drop(self, cascade: bool = False) -> str:
_cascade: str = "cascade" if cascade else ""
return reduce_stm(
params.bs_stm.drop.get(self.func_type, "func").format(
func_name=self.func_name, cascade=_cascade
)
)

# [x] Migrate to modern style
def get_func_stm_create(self) -> str:
return self.func_profile["statement"]

# [x] Migrate to modern style
def get_func_catalog(self, config: Optional[dict] = None):
"""Get merge configuration from any properties of function :structure:
<catalog_name: `{body_...}`>:
version: <iso-format: [`%Y-%m-%d`, `%Y-%m-%d %H:%M:%S`]>
description: ""
<create/statement/function>: ""
"""
_config: dict = config or get_config(
config_name=self.func_name,
config_prefix="",
folder_config=params.map_func_types.get(self.func_type, "function"),
config_prefix_file=self.func_type,
)
_config_keys: list = list(_config)
if not (
_create_key := only_one(
_config_keys, params.map_func.create, default=False
)
):
raise CatalogNotFound(
f"Catalog does not found any key represent `create`/`function`/`statement` of {self.func_name!r}"
)
self.func_catalog: dict = {
"description": _config.get(
only_one(_config_keys, params.map_func.desc), ""
),
"profile": self._generate_profile(_config.pop(_create_key)),
}
return self.func_catalog

# [x] Migrate to modern style
def _generate_profile(self, profiles: Union[dict, str]):
if isinstance(profiles, str):
return {
"statement": Statement(profiles).generate(),
"parameter": [],
}
elif isinstance(profiles, dict):
_profiles: dict = {
"parameter": list(
set(
profiles.get(
only_one(list(profiles), params.map_func.param), []
)
)
),
}
if not (_stm_key := only_one(list(profiles), params.map_func.stm)):
raise CatalogArgumentError(
f"Function profile of {self.func_name!r} does not "
f"set statement while profile key was dict type"
)
_profiles["statement"] = Statement(
profiles.get(_stm_key, "")
).generate()
return _profiles
raise CatalogArgumentError(
f"AI framework function does not support for create "
f"type {type(profiles)!r} that set in {self.func_name!r}"
)


class PipeCatalog:
"""Pipeline Catalog for generate all configuration from .yaml file."""

Expand Down
Loading

0 comments on commit 8bfd413

Please sign in to comment.