From e762ef80c907dac6232fb93e027b60d1e044f25a Mon Sep 17 00:00:00 2001 From: Tony Meyer Date: Fri, 6 Oct 2023 17:14:43 +1300 Subject: [PATCH] chore: remove as much 'if typing.TYPE_CHECKING:' as possible (#1034) * Wherever possible, remove the 'if typing.TYPE_CHECKING:' guards. * Move the definition of the `_ConfigOption` `TypedDict` from `ops.testing` to `ops.model` - `model` should not import from `testing` (the import loop was avoided because it was only when `TYPE_CHECKING` but it's still poor practice) This leaves a few cases where we still use `TYPE_CHECKING`: * When using `typing_extensions`. We currently assume that if `TYPE_CHECKING` is true that the `requirements-dev` have been installed (or at least `typing_extensions` has). It seems like this would not always be the case (e.g. if type checking is being done on a charm that uses `ops`) but is the status-quo and avoids bringing `typing_extensions` into `requirements.txt`. Hopefully, we won't need it for anything after `Required`/`NotRequired` and once we have progressed our minimum Python version enough to get those, we can drop it entirely. * We override the definition of some class attributes to be properties when `TYPE_CHECKING` is true. Removing the guard would change the actual attributes. * In Python 3.8 we can't use `weakref.WeakValueDictionary[T]` with a type. We're currently assuming that a more recent version of Python will be used when type checking, but can't bring this out as that would raise the minimum Python version for all cases (or we'd need to adjust the typing so that it can work in 3.8). --- docs/conf.py | 14 ++++---- ops/charm.py | 26 ++++++++------ ops/framework.py | 73 ++++++++++++++++++------------------- ops/log.py | 15 ++++---- ops/main.py | 21 +++++------ ops/model.py | 68 ++++++++++++++++++----------------- ops/pebble.py | 93 +++++++++++++++++++++++++++--------------------- ops/storage.py | 18 +++++----- ops/testing.py | 63 ++++++++++++++------------------ 9 files changed, 196 insertions(+), 195 deletions(-) diff --git a/docs/conf.py b/docs/conf.py index f335d64f5..b7b4c7cf0 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -73,23 +73,25 @@ def _compute_navigation_tree(context): # domain name if present. Example entries would be ('py:func', 'int') or # ('envvar', 'LD_LIBRARY_PATH'). nitpick_ignore = [ - ('py:class', '_AddressDict'), + ('py:class', 'ops.model._AddressDict'), ('py:class', '_ChangeDict'), ('py:class', '_CheckInfoDict'), + ('py:class', 'ops.model._ConfigOption'), + ('py:class', 'ops.pebble._FileLikeIO'), ('py:class', '_FileInfoDict'), - ('py:class', '_IOSource'), - ('py:class', '_NetworkDict'), + ('py:class', 'ops.pebble._IOSource'), + ('py:class', 'ops.model._NetworkDict'), ('py:class', '_ProgressDict'), ('py:class', '_Readable'), ('py:class', '_RelationMetaDict'), ('py:class', '_ResourceMetaDict'), - ('py:class', '_ServiceInfoDict'), + ('py:class', 'ops.pebble._ServiceInfoDict'), ('py:class', '_StorageMetaDict'), - ('py:class', '_SystemInfoDict'), + ('py:class', 'ops.pebble._SystemInfoDict'), ('py:class', '_TaskDict'), ('py:class', '_TextOrBinaryIO'), ('py:class', '_WarningDict'), - ('py:class', '_WebSocket'), + ('py:class', 'ops.pebble._WebSocket'), ('py:class', '_Writeable'), ('py:class', 'ops.model._ModelBackend'), ('py:class', 'ops.model._ModelCache'), diff --git a/ops/charm.py b/ops/charm.py index 15271c4db..e2b39fa06 100755 --- a/ops/charm.py +++ b/ops/charm.py @@ -35,14 +35,18 @@ from ops import model from ops._private import yaml -from ops.framework import EventBase, EventSource, Framework, Object, ObjectEvents +from ops.framework import ( + EventBase, + EventSource, + Framework, + Handle, + Object, + ObjectEvents, +) if TYPE_CHECKING: from typing_extensions import Required - from ops.framework import Handle - from ops.model import Container, Relation, Storage - _Scopes = Literal['global', 'container'] _RelationMetaDict = TypedDict( '_RelationMetaDict', { @@ -369,7 +373,7 @@ class RelationEvent(HookEvent): relations with the same name. """ - relation: 'Relation' + relation: 'model.Relation' """The relation involved in this event.""" # TODO(benhoyt): I *think* app should never be None, but confirm and update type @@ -383,7 +387,7 @@ class RelationEvent(HookEvent): :class:`Application `-level event. """ - def __init__(self, handle: 'Handle', relation: 'Relation', + def __init__(self, handle: 'Handle', relation: 'model.Relation', app: Optional[model.Application] = None, unit: Optional[model.Unit] = None): super().__init__(handle) @@ -498,7 +502,7 @@ class RelationDepartedEvent(RelationEvent): relation, the unit agent will fire the :class:`RelationBrokenEvent`. """ - def __init__(self, handle: 'Handle', relation: 'Relation', + def __init__(self, handle: 'Handle', relation: 'model.Relation', app: Optional[model.Application] = None, unit: Optional[model.Unit] = None, departing_unit_name: Optional[str] = None): @@ -563,10 +567,10 @@ class StorageEvent(HookEvent): of :class:`StorageEvent`. """ - storage: 'Storage' + storage: 'model.Storage' """Storage instance this event refers to.""" - def __init__(self, handle: 'Handle', storage: 'Storage'): + def __init__(self, handle: 'Handle', storage: 'model.Storage'): super().__init__(handle) self.storage = storage @@ -645,7 +649,7 @@ class WorkloadEvent(HookEvent): a :class:`PebbleReadyEvent`. """ - workload: 'Container' + workload: 'model.Container' """The workload involved in this event. Workload currently only can be a :class:`Container `, but @@ -653,7 +657,7 @@ class WorkloadEvent(HookEvent): for example a machine. """ - def __init__(self, handle: 'Handle', workload: 'Container'): + def __init__(self, handle: 'Handle', workload: 'model.Container'): super().__init__(handle) self.workload = workload diff --git a/ops/framework.py b/ops/framework.py index f2488f40a..54cf85778 100755 --- a/ops/framework.py +++ b/ops/framework.py @@ -37,7 +37,9 @@ Hashable, Iterable, List, + Literal, Optional, + Protocol, Set, Tuple, Type, @@ -46,6 +48,7 @@ ) from ops import charm +from ops.model import Model, _ModelBackend from ops.storage import JujuStorage, NoSnapshotError, SQLiteStorage @@ -62,23 +65,18 @@ def snapshot(self) -> Dict[str, Any]: ... # noqa def restore(self, snapshot: Dict[str, Any]) -> None: ... # noqa -if TYPE_CHECKING: - from typing import Literal, Protocol +class _StoredObject(Protocol): + _under: Any = None # noqa - from ops.charm import CharmMeta - from ops.model import Model, _ModelBackend - class _StoredObject(Protocol): - _under: Any = None # noqa +StoredObject = Union['StoredList', 'StoredSet', 'StoredDict'] - StoredObject = Union['StoredList', 'StoredSet', 'StoredDict'] - - _Path = _Kind = _MethodName = _EventKey = str - # used to type Framework Attributes - _ObserverPath = List[Tuple[_Path, _MethodName, _Path, _EventKey]] - _ObjectPath = Tuple[Optional[_Path], _Kind] - _PathToObjectMapping = Dict[_Path, 'Object'] - _PathToSerializableMapping = Dict[_Path, Serializable] +_Path = _Kind = _MethodName = _EventKey = str +# used to type Framework Attributes +_ObserverPath = List[Tuple[_Path, _MethodName, _Path, _EventKey]] +_ObjectPath = Tuple[Optional[_Path], _Kind] +_PathToObjectMapping = Dict[_Path, 'Object'] +_PathToSerializableMapping = Dict[_Path, Serializable] _T = TypeVar("_T") _EventType = TypeVar('_EventType', bound='EventBase') @@ -561,21 +559,22 @@ class Framework(Object): model: 'Model' = None # type: ignore """The :class:`Model` instance for this charm.""" - meta: 'CharmMeta' = None # type: ignore + meta: 'charm.CharmMeta' = None # type: ignore """The charm's metadata.""" charm_dir: 'pathlib.Path' = None # type: ignore """The charm project root directory.""" + _stored: 'StoredStateData' = None # type: ignore + # to help the type checker and IDEs: if TYPE_CHECKING: - _stored: 'StoredStateData' = None # type: ignore @property def on(self) -> 'FrameworkEvents': ... # noqa def __init__(self, storage: Union[SQLiteStorage, JujuStorage], charm_dir: Union[str, pathlib.Path], - meta: 'CharmMeta', model: 'Model', + meta: 'charm.CharmMeta', model: 'Model', event_name: Optional[str] = None): super().__init__(self, None) @@ -1059,14 +1058,13 @@ def __init__(self, parent: Object, attr_name: str): parent.framework.observe(parent.framework.on.commit, self._data.on_commit) # type: ignore - if TYPE_CHECKING: - @typing.overload - def __getattr__(self, key: Literal['on']) -> ObjectEvents: - pass + @typing.overload + def __getattr__(self, key: Literal['on']) -> ObjectEvents: + pass - @typing.overload - def __getattr__(self, key: str) -> Any: - pass + @typing.overload + def __getattr__(self, key: str) -> Any: + pass def __getattr__(self, key: str) -> Any: # "on" is the only reserved key that can't be used in the data map. @@ -1125,20 +1123,19 @@ def __init__(self): self.parent_type: Optional[Type[Any]] = None self.attr_name: Optional[str] = None - if TYPE_CHECKING: - @typing.overload - def __get__( - self, - parent: Literal[None], - parent_type: 'Type[_ObjectType]') -> 'StoredState': - pass - - @typing.overload - def __get__( - self, - parent: '_ObjectType', - parent_type: 'Type[_ObjectType]') -> BoundStoredState: - pass + @typing.overload + def __get__( + self, + parent: Literal[None], + parent_type: 'Type[_ObjectType]') -> 'StoredState': + pass + + @typing.overload + def __get__( + self, + parent: '_ObjectType', + parent_type: 'Type[_ObjectType]') -> BoundStoredState: + pass def __get__(self, parent: Optional['_ObjectType'], diff --git a/ops/log.py b/ops/log.py index f16bfaf38..e167c6d1a 100644 --- a/ops/log.py +++ b/ops/log.py @@ -16,19 +16,16 @@ import logging import sys +import types import typing -if typing.TYPE_CHECKING: - from types import TracebackType - from typing import Type - - from ops.model import _ModelBackend +from ops.model import _ModelBackend class JujuLogHandler(logging.Handler): """A handler for sending logs to Juju via juju-log.""" - def __init__(self, model_backend: "_ModelBackend", level: int = logging.DEBUG): + def __init__(self, model_backend: _ModelBackend, level: int = logging.DEBUG): super().__init__(level) self.model_backend = model_backend @@ -41,7 +38,7 @@ def emit(self, record: logging.LogRecord): self.model_backend.juju_log(record.levelname, self.format(record)) -def setup_root_logging(model_backend: "_ModelBackend", debug: bool = False): +def setup_root_logging(model_backend: _ModelBackend, debug: bool = False): """Setup python logging to forward messages to juju-log. By default, logging is set to DEBUG level, and messages will be filtered by Juju. @@ -62,7 +59,9 @@ def setup_root_logging(model_backend: "_ModelBackend", debug: bool = False): handler.setFormatter(formatter) logger.addHandler(handler) - def except_hook(etype: "Type[BaseException]", value: BaseException, tb: "TracebackType"): + def except_hook(etype: typing.Type[BaseException], + value: BaseException, + tb: types.TracebackType): logger.error( "Uncaught exception while in charm code:", exc_info=(etype, value, tb)) diff --git a/ops/main.py b/ops/main.py index 137965363..40948d44f 100755 --- a/ops/main.py +++ b/ops/main.py @@ -26,7 +26,7 @@ import sys import warnings from pathlib import Path -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type, Union, cast +from typing import Any, Dict, List, Optional, Tuple, Type, Union, cast import ops.charm import ops.framework @@ -36,11 +36,6 @@ from ops.jujuversion import JujuVersion from ops.log import setup_root_logging -if TYPE_CHECKING: - from ops.charm import CharmBase - from ops.framework import BoundEvent, EventSource - from ops.model import Relation - CHARM_STATE_FILE = '.unit-state.db' @@ -68,7 +63,7 @@ def _get_charm_dir(): return charm_dir -def _create_event_link(charm: 'CharmBase', bound_event: 'EventSource', +def _create_event_link(charm: 'ops.charm.CharmBase', bound_event: 'ops.framework.EventSource', link_to: Union[str, Path]): """Create a symlink for a particular event. @@ -106,7 +101,7 @@ def _create_event_link(charm: 'CharmBase', bound_event: 'EventSource', event_path.symlink_to(target_path) -def _setup_event_links(charm_dir: Path, charm: 'CharmBase'): +def _setup_event_links(charm_dir: Path, charm: 'ops.charm.CharmBase'): """Set up links for supported events that originate from Juju. Whether a charm can handle an event or not can be determined by @@ -128,7 +123,7 @@ def _setup_event_links(charm_dir: Path, charm: 'CharmBase'): _create_event_link(charm, bound_event, link_to) -def _emit_charm_event(charm: 'CharmBase', event_name: str): +def _emit_charm_event(charm: 'ops.charm.CharmBase', event_name: str): """Emits a charm event based on a Juju event name. Args: @@ -149,8 +144,8 @@ def _emit_charm_event(charm: 'CharmBase', event_name: str): event_to_emit.emit(*args, **kwargs) -def _get_event_args(charm: 'CharmBase', - bound_event: 'BoundEvent') -> Tuple[List[Any], Dict[str, Any]]: +def _get_event_args(charm: 'ops.charm.CharmBase', + bound_event: 'ops.framework.BoundEvent') -> Tuple[List[Any], Dict[str, Any]]: event_type = bound_event.event_type model = charm.framework.model @@ -189,7 +184,7 @@ def _get_event_args(charm: 'CharmBase', elif issubclass(event_type, ops.charm.RelationEvent): relation_name = os.environ['JUJU_RELATION'] relation_id = int(os.environ['JUJU_RELATION_ID'].split(':')[-1]) - relation: Optional[Relation] = model.get_relation(relation_name, relation_id) + relation: Optional[ops.model.Relation] = model.get_relation(relation_name, relation_id) remote_app_name = os.environ.get('JUJU_REMOTE_APP', '') remote_unit_name = os.environ.get('JUJU_REMOTE_UNIT', '') @@ -239,7 +234,7 @@ def __init__(self, charm_dir: Path): else: self._init_legacy() - def ensure_event_links(self, charm: 'CharmBase'): + def ensure_event_links(self, charm: 'ops.charm.CharmBase'): """Make sure necessary symlinks are present on disk.""" if self.is_dispatch_aware: # links aren't needed diff --git a/ops/model.py b/ops/model.py index 927081bc5..1e8206ab8 100644 --- a/ops/model.py +++ b/ops/model.py @@ -40,6 +40,7 @@ Generator, Iterable, List, + Literal, Mapping, MutableMapping, Optional, @@ -60,38 +61,41 @@ # a k8s spec is a mapping from names/"types" to json/yaml spec objects K8sSpec = Mapping[str, Any] -if typing.TYPE_CHECKING: - from ops.testing import _ConfigOption - - _StorageDictType = Dict[str, Optional[List['Storage']]] - _BindingDictType = Dict[Union[str, 'Relation'], 'Binding'] - - _StatusDict = TypedDict('_StatusDict', {'status': str, 'message': str}) - - # mapping from relation name to a list of relation objects - _RelationMapping_Raw = Dict[str, Optional[List['Relation']]] - # mapping from container name to container metadata - _ContainerMeta_Raw = Dict[str, ops.charm.ContainerMeta] - - # relation data is a string key: string value mapping so far as the - # controller is concerned - _RelationDataContent_Raw = Dict[str, str] - UnitOrApplicationType = Union[Type['Unit'], Type['Application']] - - _AddressDict = TypedDict('_AddressDict', { - 'address': str, # Juju < 2.9 - 'value': str, # Juju >= 2.9 - 'cidr': str - }) - _BindAddressDict = TypedDict('_BindAddressDict', { - 'interface-name': str, - 'addresses': List[_AddressDict] - }) - _NetworkDict = TypedDict('_NetworkDict', { - 'bind-addresses': List[_BindAddressDict], - 'ingress-addresses': List[str], - 'egress-subnets': List[str] - }) +_ConfigOption = TypedDict('_ConfigOption', { + 'type': Literal['string', 'int', 'float', 'boolean'], + 'description': str, + 'default': Union[str, int, float, bool], +}) + +_StorageDictType = Dict[str, Optional[List['Storage']]] +_BindingDictType = Dict[Union[str, 'Relation'], 'Binding'] + +_StatusDict = TypedDict('_StatusDict', {'status': str, 'message': str}) + +# mapping from relation name to a list of relation objects +_RelationMapping_Raw = Dict[str, Optional[List['Relation']]] +# mapping from container name to container metadata +_ContainerMeta_Raw = Dict[str, 'ops.charm.ContainerMeta'] + +# relation data is a string key: string value mapping so far as the +# controller is concerned +_RelationDataContent_Raw = Dict[str, str] +UnitOrApplicationType = Union[Type['Unit'], Type['Application']] + +_AddressDict = TypedDict('_AddressDict', { + 'address': str, # Juju < 2.9 + 'value': str, # Juju >= 2.9 + 'cidr': str +}) +_BindAddressDict = TypedDict('_BindAddressDict', { + 'interface-name': str, + 'addresses': List[_AddressDict] +}) +_NetworkDict = TypedDict('_NetworkDict', { + 'bind-addresses': List[_BindAddressDict], + 'ingress-addresses': List[str], + 'egress-subnets': List[str] +}) logger = logging.getLogger(__name__) diff --git a/ops/pebble.py b/ops/pebble.py index 70c41b1ce..6eef4aab6 100644 --- a/ops/pebble.py +++ b/ops/pebble.py @@ -136,32 +136,56 @@ 'checks': Dict[str, CheckDict]}, total=False) -if TYPE_CHECKING: - from typing_extensions import NotRequired +_AuthDict = TypedDict('_AuthDict', + {'permissions': Optional[str], + 'user-id': Optional[int], + 'user': Optional[str], + 'group-id': Optional[int], + 'group': Optional[str], + 'path': Optional[str], + 'make-dirs': Optional[bool], + 'make-parents': Optional[bool], + }, total=False) + +_ServiceInfoDict = TypedDict('_ServiceInfoDict', + {'startup': Union['ServiceStartup', str], + 'current': Union['ServiceStatus', str], + 'name': str}) + +# Callback types for _MultiParser header and body handlers + + +class _BodyHandler(Protocol): + def __call__(self, data: bytes, done: bool = False) -> None: ... # noqa + + +_HeaderHandler = Callable[[bytes], None] + +# tempfile.NamedTemporaryFile has an odd interface because of that +# 'name' attribute, so we need to make a Protocol for it. - # callback types for _MultiParser header and body handlers - class _BodyHandler(Protocol): - def __call__(self, data: bytes, done: bool = False) -> None: ... # noqa - _HeaderHandler = Callable[[bytes], None] +class _Tempfile(Protocol): + name = '' + def write(self, data: bytes): ... # noqa + def close(self): ... # noqa - # tempfile.NamedTemporaryFile has an odd interface because of that - # 'name' attribute, so we need to make a Protocol for it. - class _Tempfile(Protocol): - name = '' - def write(self, data: bytes): ... # noqa - def close(self): ... # noqa - class _FileLikeIO(Protocol[typing.AnyStr]): # That also covers TextIO and BytesIO - def read(self, __n: int = ...) -> typing.AnyStr: ... # for BinaryIO # noqa - def write(self, __s: typing.AnyStr) -> int: ... # noqa - def __enter__(self) -> typing.IO[typing.AnyStr]: ... # noqa +class _FileLikeIO(Protocol[typing.AnyStr]): # That also covers TextIO and BytesIO + def read(self, __n: int = ...) -> typing.AnyStr: ... # for BinaryIO # noqa + def write(self, __s: typing.AnyStr) -> int: ... # noqa + def __enter__(self) -> typing.IO[typing.AnyStr]: ... # noqa - _AnyStrFileLikeIO = Union[_FileLikeIO[bytes], _FileLikeIO[str]] - _TextOrBinaryIO = Union[TextIO, BinaryIO] - _IOSource = Union[str, bytes, _AnyStrFileLikeIO] - _SystemInfoDict = TypedDict('_SystemInfoDict', {'version': str}) +_AnyStrFileLikeIO = Union[_FileLikeIO[bytes], _FileLikeIO[str]] +_TextOrBinaryIO = Union[TextIO, BinaryIO] +_IOSource = Union[str, bytes, _AnyStrFileLikeIO] + +_SystemInfoDict = TypedDict('_SystemInfoDict', {'version': str}) + +if TYPE_CHECKING: + from typing_extensions import NotRequired + _CheckInfoDict = TypedDict('_CheckInfoDict', {"name": str, "level": NotRequired[Optional[Union['CheckLevel', str]]], @@ -180,21 +204,6 @@ def __enter__(self) -> typing.IO[typing.AnyStr]: ... # noqa "group": NotRequired[Optional[str]], "type": Union['FileType', str]}) - _AuthDict = TypedDict('_AuthDict', - {'permissions': Optional[str], - 'user-id': Optional[int], - 'user': Optional[str], - 'group-id': Optional[int], - 'group': Optional[str], - 'path': Optional[str], - 'make-dirs': Optional[bool], - 'make-parents': Optional[bool], - }, total=False) - _ServiceInfoDict = TypedDict('_ServiceInfoDict', - {'startup': Union['ServiceStartup', str], - 'current': Union['ServiceStatus', str], - 'name': str}) - _ProgressDict = TypedDict('_ProgressDict', {'label': str, 'done': int, @@ -238,12 +247,14 @@ def __enter__(self) -> typing.IO[typing.AnyStr]: ... # noqa 'expire-after': str, 'repeat-after': str}) - class _WebSocket(Protocol): - def connect(self, url: str, socket: socket.socket): ... # noqa - def shutdown(self): ... # noqa - def send(self, payload: str): ... # noqa - def send_binary(self, payload: bytes): ... # noqa - def recv(self) -> Union[str, bytes]: ... # noqa + +class _WebSocket(Protocol): + def connect(self, url: str, socket: socket.socket): ... # noqa + def shutdown(self): ... # noqa + def send(self, payload: str): ... # noqa + def send_binary(self, payload: bytes): ... # noqa + def recv(self) -> Union[str, bytes]: ... # noqa + logger = logging.getLogger(__name__) diff --git a/ops/storage.py b/ops/storage.py index 16206aae0..d8ddacbd5 100755 --- a/ops/storage.py +++ b/ops/storage.py @@ -19,7 +19,6 @@ import shutil import sqlite3 import subprocess -import typing from datetime import timedelta from pathlib import Path from typing import Any, Callable, Generator, List, Optional, Tuple, Union @@ -29,16 +28,15 @@ logger = logging.getLogger() -if typing.TYPE_CHECKING: - # _Notice = Tuple[event_path, observer_path, method_name] - _Notice = Tuple[str, str, str] - _Notices = List[_Notice] +# _Notice = Tuple[event_path, observer_path, method_name] +_Notice = Tuple[str, str, str] +_Notices = List[_Notice] - # This is a function that takes a Tuple and returns a yaml node. - # it replaces a method, so the first argument passed to the function - # (Any) is 'self'. - _TupleRepresenterType = Callable[[Any, Tuple[Any, ...]], yaml.Node] - _NoticeGenerator = Generator['_Notice', None, None] +# This is a function that takes a Tuple and returns a yaml node. +# it replaces a method, so the first argument passed to the function +# (Any) is 'self'. +_TupleRepresenterType = Callable[[Any, Tuple[Any, ...]], yaml.Node] +_NoticeGenerator = Generator['_Notice', None, None] def _run(args: List[str], **kw: Any): diff --git a/ops/testing.py b/ops/testing.py index 0229b74d4..57a0c52a0 100755 --- a/ops/testing.py +++ b/ops/testing.py @@ -33,7 +33,6 @@ from io import BytesIO, IOBase, StringIO from textwrap import dedent from typing import ( - TYPE_CHECKING, Any, AnyStr, BinaryIO, @@ -59,39 +58,31 @@ from ops import charm, framework, model, pebble, storage from ops._private import yaml from ops.charm import CharmBase, CharmMeta, RelationRole -from ops.model import Container, RelationNotFoundError +from ops.model import Container, RelationNotFoundError, _ConfigOption, _NetworkDict from ops.pebble import ExecProcess -if TYPE_CHECKING: - from ops.model import _NetworkDict - - ReadableBuffer = Union[bytes, str, StringIO, BytesIO, BinaryIO] - _StringOrPath = Union[str, pathlib.PurePosixPath, pathlib.Path] - _FileKwargs = TypedDict('_FileKwargs', { - 'permissions': Optional[int], - 'last_modified': datetime.datetime, - 'user_id': Optional[int], - 'user': Optional[str], - 'group_id': Optional[int], - 'group': Optional[str], - }) - - _RelationEntities = TypedDict('_RelationEntities', { - 'app': str, - 'units': List[str] - }) - - _ConfigOption = TypedDict('_ConfigOption', { - 'type': Literal['string', 'int', 'float', 'boolean'], - 'description': str, - 'default': Union[str, int, float, bool], - }) - _StatusName = Literal['unknown', 'blocked', 'active', 'maintenance', 'waiting'] - _RawStatus = TypedDict('_RawStatus', { - 'status': _StatusName, - 'message': str, - }) - RawConfig = TypedDict("RawConfig", {'options': Dict[str, _ConfigOption]}) +ReadableBuffer = Union[bytes, str, StringIO, BytesIO, BinaryIO] +_StringOrPath = Union[str, pathlib.PurePosixPath, pathlib.Path] +_FileKwargs = TypedDict('_FileKwargs', { + 'permissions': Optional[int], + 'last_modified': datetime.datetime, + 'user_id': Optional[int], + 'user': Optional[str], + 'group_id': Optional[int], + 'group': Optional[str], +}) + +_RelationEntities = TypedDict('_RelationEntities', { + 'app': str, + 'units': List[str] +}) + +_StatusName = Literal['unknown', 'blocked', 'active', 'maintenance', 'waiting'] +_RawStatus = TypedDict('_RawStatus', { + 'status': _StatusName, + 'message': str, +}) +_RawConfig = TypedDict("_RawConfig", {'options': Dict[str, _ConfigOption]}) # YAMLStringOrFile is something like metadata.yaml or actions.yaml. You can @@ -547,7 +538,7 @@ def _get_config(self, charm_config_yaml: Optional['YAMLStringOrFile']): if not isinstance(config, dict): 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: @@ -1803,7 +1794,7 @@ class _TestingConfig(Dict[str, Union[str, int, float, bool]]): 'float': float } - def __init__(self, config: 'RawConfig'): + def __init__(self, config: '_RawConfig'): super().__init__() self._spec = config self._defaults = self._load_defaults(config) @@ -1814,7 +1805,7 @@ def __init__(self, config: 'RawConfig'): self._config_set(key, value) @staticmethod - def _load_defaults(charm_config: 'RawConfig') -> Dict[str, Union[str, int, float, bool]]: + def _load_defaults(charm_config: '_RawConfig') -> Dict[str, Union[str, int, float, bool]]: """Load default values from config.yaml. Handle the case where a user doesn't supply explicit config snippets. @@ -1901,7 +1892,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