| Server IP : 74.208.127.88 / Your IP : 216.73.216.185 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /proc/self/root/snap/core20/current/usr/share/subiquity/subiquitycore/ |
Upload File : |
# Copyright 2019 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, version 3.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
import enum
import functools
import inspect
class Status(enum.Enum):
SUCCESS = enum.auto()
FAIL = enum.auto()
WARN = enum.auto()
class Context:
"""Class to report when things start and finish.
The expected way to use this is something like:
with somecontext.child("operation"):
await long_running_operation()
but you can also call .enter() and .exit() if use as a context
manager isn't possible.
start and finish events are reported via the report_start_event and
report_finish_event methods on app.
You can override the message shown on exit by passing it to the .exit
method or by assigning to description:
with somecontext.child("operation") as context:
result = await long_running_operation()
context.description = "result was {}".format(result)
"""
def __init__(self, app, name, description, parent, level, childlevel=None):
self.app = app
self.name = name
self.description = description
self.parent = parent
self.level = level
if childlevel is None:
childlevel = level
self.childlevel = childlevel
self.data = {}
@classmethod
def new(cls, app):
return cls(app, app.project, "", None, "INFO")
def child(self, name, description="", level=None, childlevel=None):
if level is None:
level = self.childlevel
return type(self)(self.app, name, description, self, level, childlevel)
def full_name(self):
c = self
names = []
while c is not None:
names.append(c.name)
c = c.parent
return '/'.join(reversed(names))
def enter(self, description=None):
if description is None:
description = self.description
self.app.report_start_event(self, description)
def exit(self, description=None, result=Status.SUCCESS):
if description is None:
description = self.description
self.app.report_finish_event(self, description, result)
def __enter__(self):
self.enter()
return self
def __exit__(self, exc, value, tb):
if exc is not None:
result = Status.FAIL
if isinstance(value, asyncio.CancelledError):
description = "cancelled"
else:
description = str(value)
else:
result = Status.SUCCESS
description = None
self.exit(description, result)
def set(self, key, value):
self.data[key] = value
def get(self, key, default=None):
c = self
while c is not None:
if key in c.data:
return c.data[key]
c = c.parent
return default
def with_context(name=None, description="", **context_kw):
def decorate(meth):
nonlocal name
if name is None:
name = meth.__name__
def convargs(self, kw):
context = kw.get('context')
if context is None:
context = self.context
kw['context'] = context.child(
name=name.format(**kw),
description=description.format(**kw),
**context_kw)
return kw
@functools.wraps(meth)
def decorated_sync(self, **kw):
kw = convargs(self, kw)
with kw['context']:
return meth(self, **kw)
@functools.wraps(meth)
async def decorated_async(self, **kw):
kw = convargs(self, kw)
with kw['context']:
return await meth(self, **kw)
if inspect.iscoroutinefunction(meth):
return decorated_async
else:
return decorated_sync
return decorate