_check_instance.py 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
  1. # Copyright 2021-2024 Avaiga Private Limited
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
  4. # the License. You may obtain a copy of the License at
  5. #
  6. # http://www.apache.org/licenses/LICENSE-2.0
  7. #
  8. # Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
  9. # an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
  10. # specific language governing permissions and limitations under the License.
  11. from typing import Union
  12. from .._entity._entity import _Entity
  13. from ..cycle.cycle import Cycle
  14. from ..data.data_node import DataNode
  15. from ..job.job import Job
  16. from ..scenario.scenario import Scenario
  17. from ..sequence.sequence import Sequence
  18. from ..submission.submission import Submission
  19. from ..task.task import Task
  20. def _is_cycle(entity: Union[_Entity, str]) -> bool:
  21. return isinstance(entity, Cycle) or (isinstance(entity, str) and entity.startswith(Cycle._ID_PREFIX))
  22. def _is_scenario(entity: Union[_Entity, str]) -> bool:
  23. return isinstance(entity, Scenario) or (isinstance(entity, str) and entity.startswith(Scenario._ID_PREFIX))
  24. def _is_sequence(entity: Union[_Entity, str]) -> bool:
  25. return isinstance(entity, Sequence) or (isinstance(entity, str) and entity.startswith(Sequence._ID_PREFIX))
  26. def _is_task(entity: Union[_Entity, str]) -> bool:
  27. return isinstance(entity, Task) or (isinstance(entity, str) and entity.startswith(Task._ID_PREFIX))
  28. def _is_job(entity: Union[_Entity, str]) -> bool:
  29. return isinstance(entity, Job) or (isinstance(entity, str) and entity.startswith(Job._ID_PREFIX))
  30. def _is_data_node(entity: Union[_Entity, str]) -> bool:
  31. return isinstance(entity, DataNode) or (isinstance(entity, str) and entity.startswith(DataNode._ID_PREFIX))
  32. def _is_submission(entity: Union[_Entity, str]) -> bool:
  33. return isinstance(entity, Submission) or (isinstance(entity, str) and entity.startswith(Submission._ID_PREFIX))