Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / codegen / inline / targets.py: 68%

25 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-11-15 07:18 +0800

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Normalize callable targets recorded by the native model build.""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Mapping 

20from typing import Any 

21 

22 

23def projected_target_path(value: Any) -> str | None: 

24 """Return the dotted target represented by a frozen observation value.""" 

25 if value is None: 

26 return None 

27 if isinstance(value, str): 

28 return value or None 

29 if isinstance(value, Mapping): 

30 target = value.get("_target_") or value.get("target") 

31 return str(target) if target else None 

32 module = getattr(value, "__module__", None) 

33 qualname = getattr(value, "__qualname__", None) 

34 return f"{module}.{qualname}" if module and qualname else None 

35 

36 

37def import_target_path(value: Any) -> str: 

38 """Return an importable ``module:qualname`` or reject an invalid target.""" 

39 target = projected_target_path(value) 

40 if not target: 

41 raise ValueError(f"snapshot injection target is not importable: {target!r}") 

42 if ":" in target: 

43 return target 

44 module_name, separator, name = target.rpartition(".") 

45 if not separator or not module_name or not name: 

46 raise ValueError(f"snapshot injection target is not importable: {target!r}") 

47 return f"{module_name}:{name}" 

48 

49 

50__all__ = ["import_target_path", "projected_target_path"]