Skip to content

refactor: introduce AbstractResource base class for resource interface #175

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
phantomii wants to merge 1 commit into
base: master
Choose a base branch
from
Open

refactor: introduce AbstractResource base class for resource interface #175

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 90 additions & 4 deletions genesis_core/elements/dm/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
# License for the specific language governing permissions and limitations
# under the License.

import abc
from functools import partial
import logging
import enum
Expand Down Expand Up @@ -506,7 +507,62 @@ class Requirement(
)


class AbstractResource(metaclass=abc.ABCMeta):

@abc.abstractmethod
def get_uri(self):
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def to_str(self, field: str) -> str:
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def get_parameter_value(self, parameter):
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def get_actual_state_safe(self):
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def render_target_state(self, engine=None):
raise NotImplementedError("Not implemented")

@property
@abc.abstractmethod
def link(self):
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def get_provider_element(self):
raise NotImplementedError("Not implemented")

@property
@abc.abstractmethod
def kind(self):
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def calculate_full_hash(self):
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def actualize(self):
raise NotImplementedError("Not implemented")

@abc.abstractmethod
def delete(self, session=None):
raise NotImplementedError("Not implemented")

@property
@abc.abstractmethod
def original(self):
raise NotImplementedError("Not implemented")
Comment on lines +510 to +561

@gemini-code-assist gemini-code-assist Bot Oct 22, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To improve type safety and align with the PR's goal, it would be beneficial to add type hints to the abstract methods. This will make the interface clearer for implementers. You may need to import typing as tp and use forward references (string quotes) for types defined later in the file or for circular dependencies.

class AbstractResource(metaclass=abc.ABCMeta):

    @abc.abstractmethod
    def get_uri(self) -> str:
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def to_str(self, field: str) -> str:
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def get_parameter_value(self, parameter: str) -> 'tp.Any':
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def get_actual_state_safe(self) -> dict:
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def render_target_state(self, engine: 'ElementEngine' | None = None) -> dict:
        raise NotImplementedError("Not implemented")

    @property
    @abc.abstractmethod
    def link(self) -> str:
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def get_provider_element(self) -> 'Element':
        raise NotImplementedError("Not implemented")

    @property
    @abc.abstractmethod
    def kind(self) -> str:
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def calculate_full_hash(self) -> str:
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def actualize(self) -> None:
        raise NotImplementedError("Not implemented")

    @abc.abstractmethod
    def delete(self, session: 'tp.Any' | None = None) -> None:
        raise NotImplementedError("Not implemented")

    @property
    @abc.abstractmethod
    def original(self) -> 'AbstractResource':
        raise NotImplementedError("Not implemented")



class Resource(
AbstractResource,
models.ModelWithUUID,
models.ModelWithTimestamp,
models.CustomPropertiesMixin,
Expand Down Expand Up @@ -891,16 +947,14 @@ def link(self):
return f"{self.element.link}.imports.${self.name}"


class ImportedResource:
class ImportedResource(AbstractResource):

def __init__(self, element, resource, name):
super().__init__()
self._element = element
self._resource = resource
self._name = name

def __getattr__(self, name):
return getattr(self._resource, name)

def get_parameter_value(self, parameter):
return type(self._resource).get_parameter_value(self, parameter)
Comment on lines 958 to 959

@gemini-code-assist gemini-code-assist Bot Oct 22, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

This implementation is broken after removing __getattr__. The call type(self._resource).get_parameter_value(self, parameter) will execute Resource.get_parameter_value on an ImportedResource instance (self). This will raise an AttributeError because ImportedResource does not have attributes like actual_resource or __allowed_methods_from_manifest__ which are accessed within that method.

A simple delegation to self._resource.get_parameter_value(parameter) is also incorrect because the resource name check inside it would use the original resource's name instead of the imported resource's name.

I suggest reconstructing the parameter string to use the original resource's name and then delegating the call. This correctly reuses the logic without code duplication and fixes the bug.

Suggested change
def get_parameter_value(self, parameter):
return type(self._resource).get_parameter_value(self, parameter)
def get_parameter_value(self, parameter):
parts = parameter.split(":")
resource_name = parts[0][1:]
if resource_name != self.name:
raise ValueError(
f"Resource name `{resource_name}` does not match the"
f" current resource name `{self.name}`"
)
new_parameter = f"${self._resource.name}:{':'.join(parts[1:])}"
return self._resource.get_parameter_value(new_parameter)


Expand All @@ -916,6 +970,38 @@ def name(self):
def link(self):
return f"{self.element.link}.imports.${self.name}"

def get_uri(self):
return self._resource.get_uri()

def to_str(self, field: str) -> str:
return self._resource.to_str(field)

def get_actual_state_safe(self):
return self._resource.get_actual_state_safe()

def render_target_state(self, engine=None):
return self._resource.render_target_state(engine)

def get_provider_element(self):
return self._resource.get_provider_element()

@property
def kind(self):
return self._resource.kind

def calculate_full_hash(self):
return self._resource.calculate_full_hash()

def actualize(self):
return self._resource.actualize()

def delete(self, session=None):
return self._resource.delete(session)

@property
def original(self):
return self._resource
Comment on lines +973 to +1003

@gemini-code-assist gemini-code-assist Bot Oct 22, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

For consistency and improved type safety, consider adding type hints to these delegated methods, similar to how to_str is typed. This will make the ImportedResource class more robust and easier to use correctly. You may need to import typing as tp.

    def get_uri(self) -> str:
        return self._resource.get_uri()

    def to_str(self, field: str) -> str:
        return self._resource.to_str(field)

    def get_actual_state_safe(self) -> dict:
        return self._resource.get_actual_state_safe()

    def render_target_state(self, engine: 'ElementEngine' | None = None) -> dict:
        return self._resource.render_target_state(engine)

    def get_provider_element(self) -> 'Element':
        return self._resource.get_provider_element()

    @property
    def kind(self) -> str:
        return self._resource.kind

    def calculate_full_hash(self) -> str:
        return self._resource.calculate_full_hash()

    def actualize(self) -> None:
        return self._resource.actualize()

    def delete(self, session: 'tp.Any' | None = None) -> None:
        return self._resource.delete(session)

    @property
    def original(self) -> 'Resource':
        return self._resource



class OutdatedResources(models.ModelWithUUID, orm.SQLStorableMixin):
__tablename__ = "em_outdated_resources_view"
Expand Down
Loading