1
0
mirror of https://github.com/dkmstr/openuds.git synced 2025-01-08 21:18:00 +03:00

Refactor code to remove unused imports and fix formatting

This commit is contained in:
Adolfo Gómez García 2024-09-07 23:47:56 +02:00
parent 3f8797f4d0
commit 6a0244e83d
No known key found for this signature in database
GPG Key ID: DD1ABF20724CDA23
2 changed files with 13 additions and 16 deletions

View File

@ -79,7 +79,6 @@ def classproperty(func: collections.abc.Callable[..., typing.Any]) -> ClassPrope
return ClassPropertyDescriptor(func) return ClassPropertyDescriptor(func)
def deprecated(func: collections.abc.Callable[P, T]) -> collections.abc.Callable[P, T]: def deprecated(func: collections.abc.Callable[P, T]) -> collections.abc.Callable[P, T]:
"""This is a decorator which can be used to mark functions """This is a decorator which can be used to mark functions
as deprecated. It will result in a warning being emitted as deprecated. It will result in a warning being emitted
@ -138,15 +137,17 @@ def deprecated_class_value(new_var_name: str) -> collections.abc.Callable[..., t
return functools.partial(innerDeprecated, newVarName=new_var_name) return functools.partial(innerDeprecated, newVarName=new_var_name)
# Keep this, but mypy does not likes it... it's perfect with pyright # Keep this, but mypy does not likes it... it's perfect with pyright
# # So only classes that have a "connect" method can use this decorator # # So only classes that have a "connect" method can use this decorator
class _HasConnect(typing.Protocol): class _HasConnect(typing.Protocol):
def connect(self) -> None: def connect(self) -> None: ...
...
# HasConnect = typing.TypeVar('HasConnect', bound=_HasConnect) # HasConnect = typing.TypeVar('HasConnect', bound=_HasConnect)
# def ensure_connected(func: collections.abc.Callable[typing.Concatenate[HasConnect, P], T]) -> collections.abc.Callable[typing.Concatenate[HasConnect, P], T]: # def ensure_connected(func: collections.abc.Callable[typing.Concatenate[HasConnect, P], T]) -> collections.abc.Callable[typing.Concatenate[HasConnect, P], T]:
def ensure_connected(func: collections.abc.Callable[P, T]) -> collections.abc.Callable[P, T]: def ensure_connected(func: collections.abc.Callable[P, T]) -> collections.abc.Callable[P, T]:
"""This decorator calls "connect" method of the class of the wrapped object""" """This decorator calls "connect" method of the class of the wrapped object"""
@ -286,8 +287,7 @@ def cached(
return CacheInfo(hits, misses, hits + misses, exec_time) return CacheInfo(hits, misses, hits + misses, exec_time)
def cache_clear() -> None: def cache_clear() -> None:
"""Clear the cache and cache statistics """Clear the cache and cache statistics"""
"""
nonlocal hits, misses, exec_time nonlocal hits, misses, exec_time
hits = misses = exec_time = 0 hits = misses = exec_time = 0
@ -345,9 +345,7 @@ def blocker(
if not GlobalConfig.BLOCK_ACTOR_FAILURES.as_bool(True) and not ignore_block_config: if not GlobalConfig.BLOCK_ACTOR_FAILURES.as_bool(True) and not ignore_block_config:
return f(*args, **kwargs) return f(*args, **kwargs)
request: typing.Optional[typing.Any] = getattr( request: typing.Optional[typing.Any] = getattr(args[0], request_attr or '_request', None)
args[0], request_attr or '_request', None
)
# No request object, so we can't block # No request object, so we can't block
if request is None or not isinstance(request, types.requests.ExtendedHttpRequest): if request is None or not isinstance(request, types.requests.ExtendedHttpRequest):
@ -416,4 +414,3 @@ def profiler(
return wrapper return wrapper
return decorator return decorator

View File

@ -287,7 +287,7 @@ class ProxmoxClient:
return False return False
return True return True
@cached('nodeNets', consts.CACHE_DURATION, args=1, kwargs=['node'], key_helper=caching_key_helper) @cached('nodeNets', consts.CACHE_DURATION, key_helper=caching_key_helper)
def get_node_networks(self, node: str, **kwargs: typing.Any) -> typing.Any: def get_node_networks(self, node: str, **kwargs: typing.Any) -> typing.Any:
return self.do_get(f'nodes/{node}/network', node=node)['data'] return self.do_get(f'nodes/{node}/network', node=node)['data']