Skip to content
New issue

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

#

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

Already on GitHub? # to your account

Typed ops.pebble.py #773

Merged
merged 6 commits into from
Jun 24, 2022
Merged

Typed ops.pebble.py #773

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
acedeed
pr comments
PietroPasotti Jun 13, 2022
9550d76
quoted literal
PietroPasotti Jun 13, 2022
e4038bb
added tests for checklevel 'raw' case and relaxed typing check
PietroPasotti Jun 13, 2022
53ce164
Merge branch 'main' into typing-pebble
jnsgruk Jun 23, 2022
dd949b8
pr comments
PietroPasotti Jun 23, 2022
93e2cfe
merge from main
PietroPasotti Jun 24, 2022
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
78 changes: 32 additions & 46 deletions ops/model.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@
from ops.jujuversion import JujuVersion

if typing.TYPE_CHECKING:
from pebble import _LayerDict # pyright: reportMissingTypeStubs=false
from typing_extensions import TypedDict

_StorageDictType = Dict[str, Optional[List['Storage']]]
Expand All @@ -73,8 +74,6 @@
_StatusDict = TypedDict('_StatusDict', {'status': str, 'message': str})

# the data structure we can use to initialize pebble layers with.
# todo: replace with pebble._LayerDict (a TypedDict) when pebble.py is typed
_LayerDict = Dict[str, '_LayerDict']
_Layer = Union[str, _LayerDict, pebble.Layer]

# mapping from relation name to a list of relation objects
Expand Down Expand Up @@ -1327,36 +1326,31 @@ def start(self, *service_names: str):
if not service_names:
raise TypeError('start expected at least 1 argument, got 0')

# fixme: remove on pebble.exec signature fix
self._pebble.start_services(service_names) # type: ignore
self._pebble.start_services(service_names)
PietroPasotti marked this conversation as resolved.
Show resolved Hide resolved

def restart(self, *service_names: str):
"""Restart the given service(s) by name."""
if not service_names:
raise TypeError('restart expected at least 1 argument, got 0')

try:
# fixme: remove on pebble.exec signature fix
self._pebble.restart_services(service_names) # type: ignore
self._pebble.restart_services(service_names)
except pebble.APIError as e:
if e.code != 400:
raise e
# support old Pebble instances that don't support the "restart" action
stop = tuple(s.name for s in self.get_services(*service_names).values(
) if s.is_running()) # type: Tuple[str, ...]
if stop:
# fixme: remove on pebble.exec signature fix
self._pebble.stop_services(stop) # type: ignore
# fixme: remove on pebble.exec signature fix
self._pebble.start_services(service_names) # type: ignore
self._pebble.stop_services(stop)
self._pebble.start_services(service_names)

def stop(self, *service_names: str):
"""Stop given service(s) by name."""
if not service_names:
raise TypeError('stop expected at least 1 argument, got 0')

# fixme: remove on pebble.exec signature fix
self._pebble.stop_services(service_names) # type: ignore
self._pebble.stop_services(service_names)

def add_layer(self, label: str, layer: '_Layer', *, combine: bool = False):
"""Dynamically add a new layer onto the Pebble configuration layers.
Expand All @@ -1372,8 +1366,7 @@ def add_layer(self, label: str, layer: '_Layer', *, combine: bool = False):
are combined into a single one considering the layer override
rules; if the layer doesn't exist, it is added as usual.
"""
# fixme: remove ignore once pebble.py is typed
self._pebble.add_layer(label, layer, combine=combine) # type: ignore
self._pebble.add_layer(label, layer, combine=combine)

def get_plan(self) -> 'pebble.Plan':
"""Get the current effective pebble configuration."""
Expand All @@ -1386,8 +1379,7 @@ def get_services(self, *service_names: str) -> '_ServiceInfoMapping':
services, otherwise return information for only the given services.
"""
names = service_names or None
# fixme: remove on pebble.exec signature fix
services = self._pebble.get_services(names) # type: ignore
services = self._pebble.get_services(names)
return ServiceInfoMapping(services)

def get_service(self, service_name: str) -> 'pebble.ServiceInfo':
Expand All @@ -1414,8 +1406,7 @@ def get_checks(
level: Optional check level to query for. If not specified, fetch
checks with any level.
"""
# fixme: remove on pebble.exec signature fix
checks = self._pebble.get_checks(names=check_names or None, level=level) # type: ignore
checks = self._pebble.get_checks(names=check_names or None, level=level)
return CheckInfoMapping(checks)

def get_check(self, check_name: str) -> 'pebble.CheckInfo':
Expand Down Expand Up @@ -1451,7 +1442,7 @@ def push(self,
source: Union[bytes, str, BinaryIO, TextIO],
*,
encoding: str = 'utf-8',
make_dirs: Optional[bool] = False,
make_dirs: bool = False,
permissions: Optional[int] = None,
user_id: Optional[int] = None,
user: Optional[str] = None,
Expand All @@ -1477,11 +1468,10 @@ def push(self,
both are specified.
"""
self._pebble.push(str(path), source, encoding=encoding,
# fixme: remove these ignores on pebble.exec signature fix
make_dirs=make_dirs, # type: ignore
permissions=permissions, # type: ignore
user_id=user_id, user=user, # type: ignore
group_id=group_id, group=group) # type: ignore
make_dirs=make_dirs,
permissions=permissions,
user_id=user_id, user=user,
group_id=group_id, group=group)

def list_files(self, path: StrOrPath, *, pattern: Optional[str] = None,
itself: bool = False) -> List['pebble.FileInfo']:
Expand All @@ -1499,8 +1489,7 @@ def list_files(self, path: StrOrPath, *, pattern: Optional[str] = None,
directory itself, rather than its contents.
"""
return self._pebble.list_files(str(path),
# fixme: remove on pebble.exec signature fix
pattern=pattern, itself=itself) # type: ignore
pattern=pattern, itself=itself)

def push_path(self,
source_path: Union[StrOrPath, Iterable[StrOrPath]],
Expand Down Expand Up @@ -1677,7 +1666,7 @@ def _build_fileinfo(path: StrOrPath) -> 'pebble.FileInfo':
name=path.name,
type=ftype,
size=info.st_size,
permissions=stat.S_IMODE(info.st_mode), # type: ignore
permissions=typing.cast(int, stat.S_IMODE(info.st_mode)), # type: ignore
last_modified=datetime.datetime.fromtimestamp(info.st_mtime),
user_id=info.st_uid,
user=pwd.getpwuid(info.st_uid).pw_name,
Expand Down Expand Up @@ -1770,11 +1759,10 @@ def make_dir(
group: Group name for directory. Group's GID must match group_id
if both are specified.
"""
# fixme: remove ignores on pebble.exec signature fix
self._pebble.make_dir(path, make_parents=make_parents,
permissions=permissions, # type: ignore
user_id=user_id, user=user, # type: ignore
group_id=group_id, group=group) # type: ignore
permissions=permissions,
user_id=user_id, user=user,
group_id=group_id, group=group)

def remove_path(self, path: str, *, recursive: bool = False):
"""Remove a file or directory on the remote system.
Expand Down Expand Up @@ -1809,19 +1797,18 @@ def exec(
"""
return self._pebble.exec(
command,
# fixme: remove ignores on pebble.py typing fix
environment=environment, # type: ignore
working_dir=working_dir, # type: ignore
timeout=timeout, # type: ignore
user_id=user_id, # type: ignore
user=user, # type: ignore
group_id=group_id, # type: ignore
group=group, # type: ignore
stdin=stdin, # type: ignore
stdout=stdout, # type: ignore
stderr=stderr, # type: ignore
encoding=encoding, # type: ignore
combine_stderr=combine_stderr, # type: ignore
environment=environment,
working_dir=working_dir,
timeout=timeout,
user_id=user_id,
user=user,
group_id=group_id,
group=group,
stdin=stdin,
stdout=stdout,
stderr=stderr,
encoding=encoding,
combine_stderr=combine_stderr,
)

def send_signal(self, sig: Union[int, str], *service_names: str):
Expand All @@ -1839,8 +1826,7 @@ def send_signal(self, sig: Union[int, str], *service_names: str):
if not service_names:
raise TypeError('send_signal expected at least 1 service name, got 0')

# fixme: remove ignore once pebble.send_signature signature is fixed
self._pebble.send_signal(sig, service_names) # type: ignore
self._pebble.send_signal(sig, service_names)


class ContainerMapping(Mapping[str, Container]):
Expand Down
Loading