Skip to content

Commit

Permalink
Refactor: Use isinstance() for type checking to enhance flexibility i… (
Browse files Browse the repository at this point in the history
#1490)

**Refactor: Use `isinstance()` for type checking to enhance flexibility
in workspace configuration**

- Replaced direct class name checks with `isinstance()` to improve
extensibility and maintainability.
- Adjusted type handling for `WorkspaceSettings`, `DockerResources`, and
`AwsResources` to ensure compatibility with future client-side
customizations.
- Streamlined object identification in `get_workspace_objects_from_file`
and related functions for better readability and reduced redundancy.
- Removed unnecessary imports and consolidated resource checks to align
with updated type-checking logic.

These changes make the workspace configuration process more robust and
adaptable to evolving client requirements.
  • Loading branch information
alphamarket authored Dec 16, 2024
1 parent 4f392b1 commit f24a819
Showing 1 changed file with 36 additions and 28 deletions.
64 changes: 36 additions & 28 deletions phi/workspace/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

def get_workspace_objects_from_file(resource_file: Path) -> dict:
"""Returns workspace objects from the resource file"""
from phi.aws.resources import AwsResources
from phi.docker.resources import DockerResources
try:
python_objects = get_python_objects_from_module(resource_file)
# logger.debug(f"python_objects: {python_objects}")
Expand All @@ -27,16 +29,18 @@ def get_workspace_objects_from_file(resource_file: Path) -> dict:
aws_resources_available = False
create_default_aws_resources = False
for obj_name, obj in python_objects.items():
_type_name = obj.__class__.__name__
if _type_name in [
"WorkspaceSettings",
"DockerResources",
"AwsResources",
]:
if isinstance(
obj,
(
WorkspaceSettings,
DockerResources,
AwsResources,
),
):
workspace_objects[obj_name] = obj
if _type_name == "DockerResources":
if isinstance(obj, DockerResources):
docker_resources_available = True
elif _type_name == "AwsResources":
elif isinstance(obj, AwsResources):
aws_resources_available = True

try:
Expand All @@ -50,7 +54,7 @@ def get_workspace_objects_from_file(resource_file: Path) -> dict:
pass

if not docker_resources_available and create_default_docker_resources:
from phi.docker.resources import DockerResources, DockerResource, DockerApp
from phi.docker.resources import DockerResource, DockerApp

logger.debug("Creating default docker resources")
default_docker_resources = DockerResources()
Expand All @@ -74,7 +78,7 @@ def get_workspace_objects_from_file(resource_file: Path) -> dict:
workspace_objects["default_docker_resources"] = default_docker_resources

if not aws_resources_available and create_default_aws_resources:
from phi.aws.resources import AwsResources, AwsResource, AwsApp
from phi.aws.resources import AwsResource, AwsApp

logger.debug("Creating default aws resources")
default_aws_resources = AwsResources()
Expand Down Expand Up @@ -171,8 +175,7 @@ def workspace_settings(self) -> Optional[WorkspaceSettings]:
try:
python_objects = get_python_objects_from_module(ws_settings_file)
for obj_name, obj in python_objects.items():
_type_name = obj.__class__.__name__
if _type_name == "WorkspaceSettings":
if isinstance(obj, WorkspaceSettings):
if self.validate_workspace_settings(obj):
self._workspace_settings = obj
if self.ws_schema is not None and self._workspace_settings is not None:
Expand Down Expand Up @@ -254,6 +257,9 @@ def get_resources(

workspace_dir_path: Optional[Path] = self.workspace_dir_path
if workspace_dir_path is not None:
from phi.aws.resources import AwsResources
from phi.docker.resources import DockerResources

logger.debug(f"--^^-- Loading workspace from: {workspace_dir_path}")
# Create a dict of objects in the workspace directory
workspace_objects = {}
Expand All @@ -274,35 +280,36 @@ def get_resources(
python_objects = get_python_objects_from_module(resource_file)
# logger.debug(f"python_objects: {python_objects}")
for obj_name, obj in python_objects.items():
_type_name = obj.__class__.__name__
if _type_name in [
"WorkspaceSettings",
"DockerResources",
"AwsResources",
]:
if isinstance(
obj,
(
WorkspaceSettings,
DockerResources,
AwsResources,
),
):
workspace_objects[obj_name] = obj
except Exception:
logger.warning(f"Error in {resource_file}")
raise

# logger.debug(f"workspace_objects: {workspace_objects}")
for obj_name, obj in workspace_objects.items():
_obj_type = obj.__class__.__name__
logger.debug(f"Loading {_obj_type}: {obj_name}")
if _obj_type == "WorkspaceSettings":
logger.debug(f"Loading {obj.__class__.__name__}: {obj_name}")
if isinstance(obj, WorkspaceSettings):
if self.validate_workspace_settings(obj):
self._workspace_settings = obj
if self.ws_schema is not None and self._workspace_settings is not None:
self._workspace_settings.ws_schema = self.ws_schema
logger.debug("Added WorkspaceSchema to WorkspaceSettings")
elif _obj_type == "DockerResources":
elif isinstance(obj, DockerResources):
if not obj.enabled:
logger.debug(f"Skipping {obj_name}: disabled")
continue
if docker_resource_groups is None:
docker_resource_groups = []
docker_resource_groups.append(obj)
elif _obj_type == "AwsResources":
elif isinstance(obj, AwsResources):
if not obj.enabled:
logger.debug(f"Skipping {obj_name}: disabled")
continue
Expand Down Expand Up @@ -365,6 +372,8 @@ def get_resources_from_file(

from sys import path as sys_path
from phi.utils.load_env import load_env
from phi.aws.resources import AwsResources
from phi.docker.resources import DockerResources

# Objects to read from the file
docker_resource_groups: Optional[List[Any]] = None
Expand All @@ -389,19 +398,18 @@ def get_resources_from_file(

# logger.debug(f"workspace_objects: {workspace_objects}")
for obj_name, obj in workspace_objects.items():
_obj_type = obj.__class__.__name__
logger.debug(f"Loading {_obj_type}: {obj_name}")
if _obj_type == "WorkspaceSettings":
logger.debug(f"Loading {obj.__class__.__module__}: {obj_name}")
if isinstance(obj, WorkspaceSettings):
if temporary_ws_config.validate_workspace_settings(obj):
temporary_ws_config._workspace_settings = obj
if _obj_type == "DockerResources":
if isinstance(obj, DockerResources):
if not obj.enabled:
logger.debug(f"Skipping {obj_name}: disabled")
continue
if docker_resource_groups is None:
docker_resource_groups = []
docker_resource_groups.append(obj)
elif _obj_type == "AwsResources":
elif isinstance(obj, AwsResources):
if not obj.enabled:
logger.debug(f"Skipping {obj_name}: disabled")
continue
Expand Down

0 comments on commit f24a819

Please sign in to comment.