From b2f6832988ba9f905ff0b983f8b1ac452166eafc Mon Sep 17 00:00:00 2001 From: Ben Hoyt Date: Mon, 16 Jan 2023 18:01:26 +1300 Subject: [PATCH] LayerDict -> _LayerDict; RawConfig -> _RawConfig; formatting fixes --- ops/charm.py | 6 +++++- ops/framework.py | 1 - ops/model.py | 4 ++-- ops/pebble.py | 24 ++++++++++++++++-------- ops/testing.py | 19 ++++++++++++------- 5 files changed, 35 insertions(+), 19 deletions(-) diff --git a/ops/charm.py b/ops/charm.py index aa00def12..37e14db67 100755 --- a/ops/charm.py +++ b/ops/charm.py @@ -34,14 +34,16 @@ from ops.framework import EventBase, EventSource, Framework, Object, ObjectEvents if TYPE_CHECKING: - from typing_extensions import Required, TypedDict from typing import Literal + from typing_extensions import Required, TypedDict + from ops.framework import Handle, JsonObject, _SerializedData from ops.model import Container, Numerical, Relation, Storage # CharmMeta also needs these. _ActionParam = Dict[str, 'JsonObject'] # + class _ActionMetaDict(TypedDict, total=False): title: str description: str @@ -49,6 +51,7 @@ class _ActionMetaDict(TypedDict, total=False): required: List[str] _Scopes = Literal['global', 'container'] + class _RelationMetaDict(TypedDict, total=False): interface: Required[str] limit: int @@ -78,6 +81,7 @@ class _PayloadMetaDict(TypedDict): class _MountDict(TypedDict, total=False): storage: Required[str] location: str + class _ContainerMetaDict(TypedDict): mounts: List[_MountDict] diff --git a/ops/framework.py b/ops/framework.py index a382a7085..5b797600d 100755 --- a/ops/framework.py +++ b/ops/framework.py @@ -50,7 +50,6 @@ if TYPE_CHECKING: from pathlib import Path - from typing import Literal, Protocol, Type from ops.charm import CharmMeta diff --git a/ops/model.py b/ops/model.py index 7ce667f55..91344976e 100644 --- a/ops/model.py +++ b/ops/model.py @@ -64,9 +64,9 @@ Client, ExecProcess, FileInfo, - LayerDict, Plan, ServiceInfo, + _LayerDict, ) from typing_extensions import TypedDict @@ -92,7 +92,7 @@ class _StatusDict(TypedDict): message: str # the data structure we can use to initialize pebble layers with. - _Layer = Union[str, LayerDict, pebble.Layer] + _Layer = Union[str, _LayerDict, pebble.Layer] # mapping from relation name to a list of relation objects _RelationMapping_Raw = Dict[str, Optional[List['Relation']]] diff --git a/ops/pebble.py b/ops/pebble.py index 449d698d4..896d944f8 100644 --- a/ops/pebble.py +++ b/ops/pebble.py @@ -62,9 +62,9 @@ if TYPE_CHECKING: from email.message import Message + from typing import Literal, Protocol from typing_extensions import TypedDict - from typing import Literal, Protocol # callback types for _MultiParser header and body handlers class _BodyHandler(Protocol): @@ -99,6 +99,7 @@ def write(self, buf: Union[bytes, str, bytearray]) -> int: ... # noqa class _SystemInfoDict(TypedDict): version: str + class _InfoDict(TypedDict): name: str level: Optional[Union['CheckLevel', str]] @@ -116,12 +117,16 @@ class _InfoDict(TypedDict): "group-id": Optional[int], "group": Optional[str], "type": Union['FileType', str]}) + class _HttpDict(TypedDict): url: str + class _TcpDict(TypedDict): port: int + class _ExecDict(TypedDict): command: str + class _CheckDict(TypedDict, total=False): override: str level: Union['CheckLevel', str] @@ -142,6 +147,7 @@ class _CheckDict(TypedDict, total=False): 'make-dirs': Optional[bool], 'make-parents': Optional[bool], }, total=False) + class _ServiceInfoDict(TypedDict): startup: Union['ServiceStartup', str] current: Union['ServiceStatus', str] @@ -200,8 +206,8 @@ class _ProgressDict(TypedDict): class _PlanDict(TypedDict, total=False): services: Dict[str, _ServiceDict] checks: Dict[str, _CheckDict] - # public as it is accessed by ops.testing - class LayerDict(TypedDict, total=False): + + class _LayerDict(TypedDict, total=False): summary: str description: str services: Dict[str, _ServiceDict] @@ -210,9 +216,11 @@ class LayerDict(TypedDict, total=False): class _Error(TypedDict): kind: str message: str + class _Item(TypedDict): path: str error: _Error + class _FilesResponse(TypedDict): result: List[_Item] @@ -704,12 +712,12 @@ class Layer: #: Mapping of check to :class:`Check` defined by this layer. checks: Dict[str, 'Check'] - def __init__(self, raw: Optional[Union[str, 'LayerDict']] = None): + def __init__(self, raw: Optional[Union[str, '_LayerDict']] = None): if isinstance(raw, str): d = yaml.safe_load(raw) or {} # type: ignore # (Any 'raw' type) else: d = raw or {} - d = typing.cast('LayerDict', d) + d = typing.cast('_LayerDict', d) self.summary = d.get('summary', '') self.description = d.get('description', '') @@ -722,7 +730,7 @@ def to_yaml(self) -> str: """Convert this layer to its YAML representation.""" return yaml.safe_dump(self.to_dict()) - def to_dict(self) -> 'LayerDict': + def to_dict(self) -> '_LayerDict': """Convert this layer to its dict representation.""" fields = [ ('summary', self.summary), @@ -731,7 +739,7 @@ def to_dict(self) -> 'LayerDict': ('checks', {name: check.to_dict() for name, check in self.checks.items()}), ] dct = {name: value for name, value in fields if value} - return typing.cast('LayerDict', dct) + return typing.cast('_LayerDict', dct) def __repr__(self) -> str: return f'Layer({self.to_dict()!r})' @@ -1728,7 +1736,7 @@ def _wait_change_using_polling(self, change_id: ChangeID, timeout: Optional[floa raise TimeoutError(f'timed out waiting for change {change_id} ({timeout} seconds)') def add_layer( - self, label: str, layer: Union[str, 'LayerDict', Layer], *, + self, label: str, layer: Union[str, '_LayerDict', Layer], *, combine: bool = False): """Dynamically add a new layer onto the Pebble configuration layers. diff --git a/ops/testing.py b/ops/testing.py index e42b070c5..999a93a45 100755 --- a/ops/testing.py +++ b/ops/testing.py @@ -56,14 +56,16 @@ from ops.model import RelationNotFoundError if TYPE_CHECKING: - from typing_extensions import TypedDict from typing import Literal + from typing_extensions import TypedDict + from ops.model import UnitOrApplication ReadableBuffer = Union[bytes, str, StringIO, BytesIO, BinaryIO] _StringOrPath = Union[str, pathlib.PurePosixPath, pathlib.Path] _FileOrDir = Union['_File', '_Directory'] + class _FileKwargs(TypedDict): permissions: int last_modified: datetime.datetime @@ -77,15 +79,18 @@ class _RelationEntities(TypedDict): units: List[str] _ConfigValue = Union[str, int, float, bool] + class _ConfigOption(TypedDict): type: Literal['string', 'int', 'float', 'boolean'] description: str default: _ConfigValue _StatusName = Literal['unknown', 'blocked', 'active', 'maintenance', 'waiting'] + class _RawStatus(TypedDict): status: _StatusName message: str - class RawConfig(TypedDict): + + class _RawConfig(TypedDict): options: Dict[str, _ConfigOption] @@ -437,7 +442,7 @@ def _get_config(self, charm_config: Optional['YAMLStringOrFile']): if not isinstance(config, dict): # pyright: reportUnnecessaryIsInstance=false raise TypeError(config) - return cast('RawConfig', config) + return cast('_RawConfig', config) def add_oci_resource(self, resource_name: str, contents: Optional[Mapping[str, str]] = None) -> None: @@ -1402,7 +1407,7 @@ class _TestingConfig(Dict[str, '_ConfigValue']): 'float': float } - def __init__(self, config: 'RawConfig'): + def __init__(self, config: '_RawConfig'): super().__init__() self._spec = config self._defaults = self._load_defaults(config) @@ -1413,7 +1418,7 @@ def __init__(self, config: 'RawConfig'): self._config_set(key, value) @staticmethod - def _load_defaults(charm_config: 'RawConfig') -> Dict[str, '_ConfigValue']: + def _load_defaults(charm_config: '_RawConfig') -> Dict[str, '_ConfigValue']: """Load default values from config.yaml. Handle the case where a user doesn't supply explicit config snippets. @@ -1500,7 +1505,7 @@ class _TestingModelBackend: as the only public methods of this type are for implementing ModelBackend. """ - def __init__(self, unit_name: str, meta: charm.CharmMeta, config: 'RawConfig'): + def __init__(self, unit_name: str, meta: charm.CharmMeta, config: '_RawConfig'): self.unit_name = unit_name self.app_name = self.unit_name.split('/')[0] self.model_name = None @@ -2190,7 +2195,7 @@ def wait_change( raise NotImplementedError(self.wait_change) def add_layer( - self, label: str, layer: Union[str, 'pebble.LayerDict', pebble.Layer], *, + self, label: str, layer: Union[str, 'pebble._LayerDict', pebble.Layer], *, combine: bool = False): # I wish we could combine some of this helpful object corralling with the actual backend, # rather than having to re-implement it. Maybe we could subclass