libervia-backend: libervia/backend/memory/memory.py comparison

comparison libervia/backend/memory/memory.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	64a85ce8be70
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 def __setitem__(self, key, value):
 raise NotImplementedError("You need do use new_session to create a session")
 def __delitem__(self, session_id):
-""" delete the session data """
+"""delete the session data"""
 self._purge_session(session_id)
 def keys(self):
 return list(self._sessions.keys())
 self.config = tools_config.parse_main_conf(log_filenames=True)
 self._cache_path = Path(self.config_get("", "local_dir"), C.CACHE_DIR)
 self.admins = self.config_get("", "admins_list", [])
 self.admin_jids = set()
 self._file_path_lock = defer.DeferredLock()
 async def initialise(self):
 self.storage = Storage()
 await self.storage.initialise()
 PersistentDict.storage = self.storage
 except RuntimeError:
 log.warning(f"Invalid JID for admin {admin}: {admin_jid_s}")
 else:
 self.admin_jids.add(admin_jid)
 ## Configuration ##
 def config_get(
-self,
+self, section: str | None, name: str, default: Any = None
-section: str|None,
+) -> str | list | dict:
-name: str,
-default: Any = None
-) -> str|list|dict:
 """Get the main configuration option
 @param section: section of the config file (None or '' for DEFAULT)
 @param name: name of the option
 @param default: value to use if not found
 return self.params.load_ind_params(profile)
 ## Profiles/Sessions management ##
 def start_session(self, password, profile):
-""""Iniatialise session for a profile
+""" "Iniatialise session for a profile
 @param password(unicode): profile session password
 or empty string is no password is set
 @param profile: %(doc_profile)s
 @raise exceptions.ProfileUnknownError if profile doesn't exists
 # if there is a value at this point in self._entities_cache,
 # it is the load_individual_params Deferred, the session is starting
 session_d = self._entities_cache[profile]
 except KeyError:
 # else we do request the params
-session_d = self._entities_cache[profile] = self.load_individual_params(
+session_d = self._entities_cache[profile] = (
-profile
+self.load_individual_params(profile)
 )
 session_d.addCallback(create_session)
 finally:
 return session_d
 sat_cipher = await self.param_get_a_async(
 C.PROFILE_PASS_PATH[1], C.PROFILE_PASS_PATH[0], profile_key=profile
 )
 valid = PasswordHasher.verify(password, sat_cipher)
 if not valid:
-log.warning(_("Authentication failure of profile {profile}").format(
+log.warning(
-profile=profile))
+_("Authentication failure of profile {profile}").format(profile=profile)
+)
 raise exceptions.PasswordError("The provided profile password doesn't match.")
 return await self.new_auth_session(password, profile)
 async def new_auth_session(self, key, profile):
 """Start a new session for the authenticated profile.
 self.params.purge_profile(profile)
 try:
 del self._entities_cache[profile]
 except KeyError:
 log.error(
-_(
+_("Trying to purge roster status cache for a profile not in memory: [%s]")
-"Trying to purge roster status cache for a profile not in memory: [%s]"
-)
 % profile
 )
 def get_profiles_list(self, clients=True, components=False):
 """retrieve profiles list
 def init_personal_key(__):
 # be sure to call this after checking that the profile doesn't exist yet
 # generated once for all and saved in a PersistentDict
-personal_key = BlockCipher.get_random_key(
+personal_key = BlockCipher.get_random_key(base64=True).decode("utf-8")
-base64=True
-).decode('utf-8')
 self.auth_sessions.new_session(
 {C.MEMORY_CRYPTO_KEY: personal_key}, profile=name
 )  # will be encrypted by param_set
 def start_fake_session(__):
 def add_to_history(self, client, data):
 return self.storage.add_to_history(data, client.profile)
 def _history_get_serialise(self, history_data):
 return [
-(uid, timestamp, from_jid, to_jid, message, subject, mess_type,
+(
-data_format.serialise(extra)) for uid, timestamp, from_jid, to_jid, message,
+uid,
-subject, mess_type, extra in history_data
+timestamp,
+from_jid,
+to_jid,
+message,
+subject,
+mess_type,
+data_format.serialise(extra),
+)
+for uid, timestamp, from_jid, to_jid, message, subject, mess_type, extra in history_data
 ]
-def _history_get(self, from_jid_s, to_jid_s, limit=C.HISTORY_LIMIT_NONE, between=True,
+def _history_get(
-filters=None, profile=C.PROF_KEY_NONE):
+self,
+from_jid_s,
+to_jid_s,
+limit=C.HISTORY_LIMIT_NONE,
+between=True,
+filters=None,
+profile=C.PROF_KEY_NONE,
+):
 from_jid = jid.JID(from_jid_s) if from_jid_s else None
 to_jid = jid.JID(to_jid_s) if to_jid_s else None
-d = self.history_get(
+d = self.history_get(from_jid, to_jid, limit, between, filters, profile)
-from_jid, to_jid, limit, between, filters, profile
-)
 d.addCallback(self._history_get_serialise)
 return d
 def history_get(
 self,
-from_jid: jid.JID|None,
+from_jid: jid.JID | None,
-to_jid: jid.JID|None,
+to_jid: jid.JID | None,
 limit: int = C.HISTORY_LIMIT_NONE,
 between: bool = True,
-filters: dict[str, str]|None = None,
+filters: dict[str, str] | None = None,
-profile: str = C.PROF_KEY_NONE
+profile: str = C.PROF_KEY_NONE,
 ) -> defer.Deferred[list]:
 """Retrieve messages in history
 @param from_jid: source JID (full, or bare for catchall)
 @param to_jid: dest JID (full, or bare for catchall)
 limit = int(self.param_get_a(C.HISTORY_LIMIT, "General", profile_key=profile))
 elif limit == C.HISTORY_LIMIT_NONE:
 limit = None
 if limit == 0:
 return defer.succeed([])
-return self.storage.history_get(from_jid, to_jid, limit, between, filters, profile)
+return self.storage.history_get(
+from_jid, to_jid, limit, between, filters, profile
+)
 ## Statuses ##
 def _get_presence_statuses(self, profile_key):
 ret = self.presence_statuses_get(profile_key)
 @param statuses: dictionary of statuses
 @param profile_key: %(doc_profile_key)s
 """
 client = self.host.get_client(profile_key)
 presence_data = PresenceTuple(show, priority, statuses)
-self.update_entity_data(
+self.update_entity_data(client, entity_jid, "presence", presence_data)
-client, entity_jid, "presence", presence_data
-)
 if entity_jid.resource and show != C.PRESENCE_UNAVAILABLE:
 # If a resource is available, bare jid should not have presence information
 try:
 self.del_entity_datum(client, entity_jid.userhostJID(), "presence")
 except (KeyError, exceptions.UnknownEntityError):
 @raise ValueError: entity_jid has a resource
 """
 # FIXME: is there a need to keep cache data for resources which are not connected anymore?
 if entity_jid.resource:
 raise ValueError(
-"get_all_resources must be used with a bare jid (got {})".format(entity_jid)
+"get_all_resources must be used with a bare jid (got {})".format(
+entity_jid
+)
 )
 profile_cache = self._get_profile_cache(client)
 try:
 entity_data = profile_cache[entity_jid.userhostJID()]
 except KeyError:
 @param entity_jid: bare entity jid
 @return (unicode): main resource or None
 """
 if entity_jid.resource:
 raise ValueError(
-"main_resource_get must be used with a bare jid (got {})".format(entity_jid)
+"main_resource_get must be used with a bare jid (got {})".format(
+entity_jid
+)
 )
 try:
 if self.host.plugins["XEP-0045"].is_joined_room(client, entity_jid):
 return None  # MUC rooms have no main resource
 except KeyError:  # plugin not found
 continue
 full_jid = copy.copy(bare_jid)
 full_jid.resource = resource
 yield full_jid
-def update_entity_data(
+def update_entity_data(self, client, entity_jid, key, value, silent=False):
-self, client, entity_jid, key, value, silent=False
-):
 """Set a misc data for an entity
 If key was registered with set_signal_on_update, a signal will be sent to frontends
 @param entity_jid: JID of the entity, C.ENTITY_ALL_RESOURCES for all resources of
 all entities, C.ENTITY_ALL for all entities (all resources + bare jids)
 )
 entity_data[key] = value
 if key in self._key_signals and not silent:
 self.host.bridge.entity_data_updated(
-jid_.full(),
+jid_.full(), key, data_format.serialise(value), client.profile
-key,
-data_format.serialise(value),
-client.profile
 )
 def del_entity_datum(self, client, entity_jid, key):
 """Delete a data for an entity
 for jid_ in entities:
 try:
 entity_data = profile_cache[jid_.userhostJID()][jid_.resource]
 except KeyError:
-raise exceptions.UnknownEntityError(
+raise exceptions.UnknownEntityError("Entity {} not in cache".format(jid_))
-"Entity {} not in cache".format(jid_)
-)
 try:
 del entity_data[key]
 except KeyError as e:
 if entity_jid in (C.ENTITY_ALL_RESOURCES, C.ENTITY_ALL):
 continue  # we ignore KeyError when deleting keys from several entities
 client = self.host.get_client(profile_key)
 ret = self.entities_data_get(
 client, [jid.JID(jid_) for jid_ in entities_jids], keys_list
 )
 return {
-jid_.full(): {k: data_format.serialise(v) for k,v in data.items()}
+jid_.full(): {k: data_format.serialise(v) for k, v in data.items()}
 for jid_, data in ret.items()
 }
 def entities_data_get(self, client, entities_jids, keys_list=None):
 """Get a list of cached values for several entities at once
 return ret_data
 def _get_entity_data(self, entity_jid_s, keys_list=None, profile=C.PROF_KEY_NONE):
 return self.entity_data_get(
-self.host.get_client(profile), jid.JID(entity_jid_s), keys_list)
+self.host.get_client(profile), jid.JID(entity_jid_s), keys_list
+)
 def entity_data_get(self, client, entity_jid, keys_list=None):
 """Get a list of cached values for entity
 @param entity_jid: JID of the entity
 profile_cache = self._get_profile_cache(client)
 try:
 entity_data = profile_cache[entity_jid.userhostJID()][entity_jid.resource]
 except KeyError:
 raise exceptions.UnknownEntityError(
-"Entity {} not in cache (was requesting {})".format(
+"Entity {} not in cache (was requesting {})".format(entity_jid, keys_list)
-entity_jid, keys_list
-)
 )
 if keys_list is None:
 return entity_data
 return {key: entity_data[key] for key in keys_list if key in entity_data}
 return self.subscriptions[profile]
 ## Parameters ##
-def get_string_param_a(self, name, category, attr="value", profile_key=C.PROF_KEY_NONE):
+def get_string_param_a(
+self, name, category, attr="value", profile_key=C.PROF_KEY_NONE
+):
 return self.params.get_string_param_a(name, category, attr, profile_key)
 def param_get_a(self, name, category, attr="value", profile_key=C.PROF_KEY_NONE):
 return self.params.param_get_a(name, category, attr, profile_key=profile_key)
 return self.params._get_params_values_from_category(
 category, security_limit, app, extra_s, profile_key
 )
 def async_get_string_param_a(
-self, name, category, attribute="value", security_limit=C.NO_SECURITY_LIMIT,
+self,
-profile_key=C.PROF_KEY_NONE):
+name,
+category,
+attribute="value",
+security_limit=C.NO_SECURITY_LIMIT,
+profile_key=C.PROF_KEY_NONE,
+):
 profile = self.get_profile_name(profile_key)
-return defer.ensureDeferred(self.params.async_get_string_param_a(
+return defer.ensureDeferred(
-name, category, attribute, security_limit, profile
+self.params.async_get_string_param_a(
-))
+name, category, attribute, security_limit, profile
+)
+)
 def _get_params_ui(self, security_limit, app, extra_s, profile_key):
 return self.params._get_params_ui(security_limit, app, extra_s, profile_key)
 def params_categories_get(self):
 def _private_data_set(self, namespace, key, data_s, profile_key):
 client = self.host.get_client(profile_key)
 # we accept any type
 data = data_format.deserialise(data_s, type_check=None)
-return defer.ensureDeferred(self.storage.set_private_value(
+return defer.ensureDeferred(
-namespace, key, data, binary=True, profile=client.profile))
+self.storage.set_private_value(
+namespace, key, data, binary=True, profile=client.profile
+)
+)
 def _private_data_get(self, namespace, key, profile_key):
 client = self.host.get_client(profile_key)
 d = defer.ensureDeferred(
 self.storage.get_privates(
-namespace, [key], binary=True, profile=client.profile)
+namespace, [key], binary=True, profile=client.profile
+)
 )
 d.addCallback(lambda data_dict: data_format.serialise(data_dict.get(key)))
 return d
 def _private_data_delete(self, namespace, key, profile_key):
 client = self.host.get_client(profile_key)
-return defer.ensureDeferred(self.storage.del_private_value(
+return defer.ensureDeferred(
-namespace, key, binary=True, profile=client.profile))
+self.storage.del_private_value(
+namespace, key, binary=True, profile=client.profile
+)
+)
 ## Files ##
 def check_file_permission(
 self,
 file_data: dict,
 peer_jid: Optional[jid.JID],
 perms_to_check: Optional[Tuple[str]],
-set_affiliation: bool = False
+set_affiliation: bool = False,
 ) -> None:
 """Check that an entity has the right permission on a file
 @param file_data: data of one file, as returned by get_files
 @param peer_jid: entity trying to access the file
 if peer_jid is None and perms_to_check is None:
 return
 peer_jid = peer_jid.userhostJID()
 if peer_jid == file_data["owner"]:
 if set_affiliation:
-file_data['affiliation'] = 'owner'
+file_data["affiliation"] = "owner"
 # the owner has all rights, nothing to check
 return
 if not C.ACCESS_PERMS.issuperset(perms_to_check):
 raise exceptions.InternalError(_("invalid permission"))
 # No permission is set.
 # If we are in a root file/directory, we deny access
 # otherwise, we use public permission, as the parent directory will
 # block anyway, this avoid to have to recursively change permissions for
 # all sub directories/files when modifying a permission
-if not file_data.get('parent'):
+if not file_data.get("parent"):
 raise exceptions.PermissionError()
 else:
 perm_type = C.ACCESS_TYPE_PUBLIC
 if perm_type == C.ACCESS_TYPE_PUBLIC:
 continue
 return (parent, [])
 def get_file_affiliations(self, file_data: dict) -> Dict[jid.JID, str]:
 """Convert file access to pubsub like affiliations"""
 affiliations = {}
-access_data = file_data['access']
+access_data = file_data["access"]
 read_data = access_data.get(C.ACCESS_PERM_READ, {})
-if read_data.get('type') == C.ACCESS_TYPE_WHITELIST:
+if read_data.get("type") == C.ACCESS_TYPE_WHITELIST:
-for entity_jid_s in read_data['jids']:
+for entity_jid_s in read_data["jids"]:
 entity_jid = jid.JID(entity_jid_s)
-affiliations[entity_jid] = 'member'
+affiliations[entity_jid] = "member"
 write_data = access_data.get(C.ACCESS_PERM_WRITE, {})
-if write_data.get('type') == C.ACCESS_TYPE_WHITELIST:
+if write_data.get("type") == C.ACCESS_TYPE_WHITELIST:
-for entity_jid_s in write_data['jids']:
+for entity_jid_s in write_data["jids"]:
 entity_jid = jid.JID(entity_jid_s)
-affiliations[entity_jid] = 'publisher'
+affiliations[entity_jid] = "publisher"
-owner = file_data.get('owner')
+owner = file_data.get("owner")
 if owner:
-affiliations[owner] = 'owner'
+affiliations[owner] = "owner"
 return affiliations
 def _set_file_affiliations_update(
-self,
+self, access: dict, file_data: dict, affiliations: Dict[jid.JID, str]
-access: dict,
-file_data: dict,
-affiliations: Dict[jid.JID, str]
 ) -> None:
 read_data = access.setdefault(C.ACCESS_PERM_READ, {})
-if read_data.get('type') != C.ACCESS_TYPE_WHITELIST:
+if read_data.get("type") != C.ACCESS_TYPE_WHITELIST:
-read_data['type'] = C.ACCESS_TYPE_WHITELIST
+read_data["type"] = C.ACCESS_TYPE_WHITELIST
-if 'jids' not in read_data:
+if "jids" not in read_data:
-read_data['jids'] = []
+read_data["jids"] = []
-read_whitelist = read_data['jids']
+read_whitelist = read_data["jids"]
 write_data = access.setdefault(C.ACCESS_PERM_WRITE, {})
-if write_data.get('type') != C.ACCESS_TYPE_WHITELIST:
+if write_data.get("type") != C.ACCESS_TYPE_WHITELIST:
-write_data['type'] = C.ACCESS_TYPE_WHITELIST
+write_data["type"] = C.ACCESS_TYPE_WHITELIST
-if 'jids' not in write_data:
+if "jids" not in write_data:
-write_data['jids'] = []
+write_data["jids"] = []
-write_whitelist = write_data['jids']
+write_whitelist = write_data["jids"]
 for entity_jid, affiliation in affiliations.items():
 entity_jid_s = entity_jid.full()
 if affiliation == "none":
 try:
 read_whitelist.remove(entity_jid_s)
 raise NotImplementedError('"owner" affiliation can\'t be set')
 else:
 raise ValueError(f"unknown affiliation: {affiliation!r}")
 async def set_file_affiliations(
-self,
+self, client, file_data: dict, affiliations: Dict[jid.JID, str]
-client,
-file_data: dict,
-affiliations: Dict[jid.JID, str]
 ) -> None:
 """Apply pubsub like affiliation to file_data
 Affiliations are converted to access types, then set in a whitelist.
 Affiliation are mapped as follow:
 - "owner" can't be set (for now)
 - "publisher" gives read and write permissions
 - "member" gives read permission only
 - "none" removes both read and write permissions
 """
-file_id = file_data['id']
+file_id = file_data["id"]
 await self.file_update(
 file_id,
-'access',
+"access",
 update_cb=partial(
 self._set_file_affiliations_update,
 file_data=file_data,
-affiliations=affiliations
+affiliations=affiliations,
 ),
 )
 def _set_file_access_model_update(
-self,
+self, access: dict, file_data: dict, access_model: str
-access: dict,
-file_data: dict,
-access_model: str
 ) -> None:
 read_data = access.setdefault(C.ACCESS_PERM_READ, {})
 if access_model == "open":
 requested_type = C.ACCESS_TYPE_PUBLIC
 elif access_model == "whitelist":
 requested_type = C.ACCESS_TYPE_WHITELIST
 else:
 raise ValueError(f"unknown access model: {access_model}")
-read_data['type'] = requested_type
+read_data["type"] = requested_type
-if requested_type == C.ACCESS_TYPE_WHITELIST and 'jids' not in read_data:
+if requested_type == C.ACCESS_TYPE_WHITELIST and "jids" not in read_data:
-read_data['jids'] = []
+read_data["jids"] = []
 async def set_file_access_model(
 self,
 client,
 file_data: dict,
 Only 2 access models are supported so far:
 - "open": set public access to file/dir
 - "whitelist": set whitelist to file/dir
 """
-file_id = file_data['id']
+file_id = file_data["id"]
 await self.file_update(
 file_id,
-'access',
+"access",
 update_cb=partial(
 self._set_file_access_model_update,
 file_data=file_data,
-access_model=access_model
+access_model=access_model,
 ),
 )
 def get_files_owner(
 self,
 client,
 owner: Optional[jid.JID],
 peer_jid: Optional[jid.JID],
 file_id: Optional[str] = None,
-parent: Optional[str] = None
+parent: Optional[str] = None,
 ) -> jid.JID:
 """Get owner to use for a file operation
 if owner is not explicitely set, a suitable one will be used (client.jid for
 clients, peer_jid for components).
 "Owner must be set for component if peer_jid is None"
 )
 return peer_jid.userhostJID()
 async def get_files(
-self, client, peer_jid, file_id=None, version=None, parent=None, path=None,
+self,
-type_=None, file_hash=None, hash_algo=None, name=None, namespace=None,
+client,
-mime_type=None, public_id=None, owner=None, access=None, projection=None,
+peer_jid,
-unique=False, perms_to_check=(C.ACCESS_PERM_READ,)):
+file_id=None,
+version=None,
+parent=None,
+path=None,
+type_=None,
+file_hash=None,
+hash_algo=None,
+name=None,
+namespace=None,
+mime_type=None,
+public_id=None,
+owner=None,
+access=None,
+projection=None,
+unique=False,
+perms_to_check=(C.ACCESS_PERM_READ,),
+):
 """Retrieve files with with given filters
 @param peer_jid(jid.JID, None): jid trying to access the file
 needed to check permission.
 Use None to ignore permission (perms_to_check must be None too)
 for file_data in to_remove:
 files.remove(file_data)
 return files
 async def set_file(
-self, client, name, file_id=None, version="", parent=None, path=None,
+self,
-type_=C.FILE_TYPE_FILE, file_hash=None, hash_algo=None, size=None,
+client,
-namespace=None, mime_type=None, public_id=None, created=None, modified=None,
+name,
-owner=None, access=None, extra=None, peer_jid=None,
+file_id=None,
-perms_to_check=(C.ACCESS_PERM_WRITE,)
+version="",
+parent=None,
+path=None,
+type_=C.FILE_TYPE_FILE,
+file_hash=None,
+hash_algo=None,
+size=None,
+namespace=None,
+mime_type=None,
+public_id=None,
+created=None,
+modified=None,
+owner=None,
+access=None,
+extra=None,
+peer_jid=None,
+perms_to_check=(C.ACCESS_PERM_WRITE,),
 ):
 """Set a file metadata
 @param name(unicode): basename of the file
 @param file_id(unicode): unique id of the file
 if mime_type is None:
 mime_type, __ = mimetypes.guess_type(name)
 else:
 mime_type = mime_type.lower()
 if public_id is not None:
-assert len(public_id)>0
+assert len(public_id) > 0
 if created is None:
 created = time.time()
 if namespace is not None:
 namespace = namespace.strip() or None
 if type_ == C.FILE_TYPE_DIRECTORY:
 access=access,
 extra=extra,
 )
 async def file_get_used_space(
-self,
+self, client, peer_jid: jid.JID, owner: Optional[jid.JID] = None
-client,
-peer_jid: jid.JID,
-owner: Optional[jid.JID] = None
 ) -> int:
 """Get space taken by all files owned by an entity
 @param peer_jid: entity requesting the size
 @param owner: entity owning the file to check. If None, will be determined by
 self,
 client,
 peer_jid: jid.JID,
 recursive: bool,
 files_path: Path,
-file_data: dict
+file_data: dict,
 ):
 """Internal method to delete files/directories recursively
 @param peer_jid(jid.JID): entity requesting the deletion (must be owner of files
 to delete)
 @param recursive(boolean): True if recursive deletion is needed
 @param files_path(unicode): path of the directory containing the actual files
 @param file_data(dict): data of the file to delete
 """
-if file_data['owner'] != peer_jid:
+if file_data["owner"] != peer_jid:
 raise exceptions.PermissionError(
-"file {file_name} can't be deleted, {peer_jid} is not the owner"
+"file {file_name} can't be deleted, {peer_jid} is not the owner".format(
-.format(file_name=file_data['name'], peer_jid=peer_jid.full()))
+file_name=file_data["name"], peer_jid=peer_jid.full()
-if file_data['type'] == C.FILE_TYPE_DIRECTORY:
+)
-sub_files = yield self.get_files(client, peer_jid, parent=file_data['id'])
+)
+if file_data["type"] == C.FILE_TYPE_DIRECTORY:
+sub_files = yield self.get_files(client, peer_jid, parent=file_data["id"])
 if sub_files and not recursive:
 raise exceptions.DataError(_("Can't delete directory, it is not empty"))
 # we first delete the sub-files
 for sub_file_data in sub_files:
-if sub_file_data['type'] == C.FILE_TYPE_DIRECTORY:
+if sub_file_data["type"] == C.FILE_TYPE_DIRECTORY:
-sub_file_path = files_path / sub_file_data['name']
+sub_file_path = files_path / sub_file_data["name"]
 else:
 sub_file_path = files_path
 yield self._delete_file(
-client, peer_jid, recursive, sub_file_path, sub_file_data)
+client, peer_jid, recursive, sub_file_path, sub_file_data
+)
 # then the directory itself
-yield self.storage.file_delete(file_data['id'])
+yield self.storage.file_delete(file_data["id"])
-elif file_data['type'] == C.FILE_TYPE_FILE:
+elif file_data["type"] == C.FILE_TYPE_FILE:
-log.info(_("deleting file {name} with hash {file_hash}").format(
+log.info(
-name=file_data['name'], file_hash=file_data['file_hash']))
+_("deleting file {name} with hash {file_hash}").format(
-yield self.storage.file_delete(file_data['id'])
+name=file_data["name"], file_hash=file_data["file_hash"]
+)
+)
+yield self.storage.file_delete(file_data["id"])
 references = yield self.get_files(
-client, peer_jid, file_hash=file_data['file_hash'])
+client, peer_jid, file_hash=file_data["file_hash"]
+)
 if references:
 log.debug("there are still references to the file, we keep it")
 else:
-file_path = os.path.join(files_path, file_data['file_hash'])
+file_path = os.path.join(files_path, file_data["file_hash"])
-log.info(_("no reference left to {file_path}, deleting").format(
+log.info(
-file_path=file_path))
+_("no reference left to {file_path}, deleting").format(
+file_path=file_path
+)
+)
 try:
 os.unlink(file_path)
 except FileNotFoundError:
-log.error(f"file at {file_path!r} doesn't exist but it was referenced in files database")
+log.error(
+f"file at {file_path!r} doesn't exist but it was referenced in files database"
+)
 else:
-raise exceptions.InternalError('Unexpected file type: {file_type}'
+raise exceptions.InternalError(
-.format(file_type=file_data['type']))
+"Unexpected file type: {file_type}".format(file_type=file_data["type"])
+)
 async def file_delete(self, client, peer_jid, file_id, recursive=False):
 """Delete a single file or a directory and all its sub-files
 @param file_id(unicode): id of the file to delete
 """
 # FIXME: we only allow owner of file to delete files for now, but WRITE access
 #        should be checked too
 files_data = await self.get_files(client, peer_jid, file_id)
 if not files_data:
-raise exceptions.NotFound("Can't find the file with id {file_id}".format(
+raise exceptions.NotFound(
-file_id=file_id))
+"Can't find the file with id {file_id}".format(file_id=file_id)
+)
 file_data = files_data[0]
 if file_data["type"] != C.FILE_TYPE_DIRECTORY and recursive:
 raise ValueError("recursive can only be set for directories")
 files_path = self.host.get_local_path(None, C.FILES_DIR)
 await self._delete_file(client, peer_jid, recursive, files_path, file_data)
 """
 namespace = namespace.strip().lower()
 return Path(
 self._cache_path,
 regex.path_escape(namespace),
-*(regex.path_escape(a) for a in args)
+*(regex.path_escape(a) for a in args),
 )
 ## Notifications ##
 def _add_notification(
 self,
 type_: str,
 body_plain: str,
 is_global: bool,
 requires_action: bool,
 priority: str,
 expire_at: float,
 extra_s: str,
-profile_key: str
+profile_key: str,
 ) -> defer.Deferred:
 client = self.host.get_client(profile_key)
 if not client.is_admin:
 raise exceptions.PermissionError("Only admins can add a notification")
 try:
 notification_type = NotificationType[type_]
 notification_priority = NotificationPriority[priority]
 except KeyError as e:
-raise exceptions.DataError(
+raise exceptions.DataError(f"invalid notification type or priority data: {e}")
-f"invalid notification type or priority data: {e}"
-)
 return defer.ensureDeferred(
 self.add_notification(
 client,
 notification_type,
 title or None,
 is_global,
 requires_action,
 notification_priority,
 expire_at or None,
-data_format.deserialise(extra_s)
+data_format.deserialise(extra_s),
 )
 )
 async def add_notification(
 self,
 @priority: how urgent the notification is
 @param expire_at: expiration timestamp for the notification.
 @param extra: additional data.
 """
 notification = await self.storage.add_notification(
-None if is_global else client, type_, body_plain, body_rich, title,
+None if is_global else client,
-requires_action, priority, expire_at, extra
+type_,
+body_plain,
+body_rich,
+title,
+requires_action,
+priority,
+expire_at,
+extra,
 )
 self.host.bridge.notification_new(
 str(notification.id),
 notification.timestamp,
 type_.value,
 body_plain,
-body_rich or '',
+body_rich or "",
-title or '',
+title or "",
 requires_action,
 priority.value,
 expire_at or 0,
-data_format.serialise(extra) if extra else '',
+data_format.serialise(extra) if extra else "",
-C.PROF_KEY_ALL if is_global else client.profile
+C.PROF_KEY_ALL if is_global else client.profile,
 )
 def _get_notifications(self, filters_s: str, profile_key: str) -> defer.Deferred:
 """Fetch notifications for bridge with given filters and profile key.
 client = self.host.get_client(profile_key)
 filters = data_format.deserialise(filters_s)
 try:
-if 'type' in filters:
+if "type" in filters:
-filters['type_'] = NotificationType[filters.pop('type')]
+filters["type_"] = NotificationType[filters.pop("type")]
-if 'status' in filters:
+if "status" in filters:
-filters['status'] = NotificationStatus[filters['status']]
+filters["status"] = NotificationStatus[filters["status"]]
-if 'min_priority' in filters:
+if "min_priority" in filters:
-filters['min_priority'] = NotificationPriority[filters['min_priority']].value
+filters["min_priority"] = NotificationPriority[
+filters["min_priority"]
+].value
 except KeyError as e:
 raise exceptions.DataError(f"invalid filter data: {e}")
 d = defer.ensureDeferred(self.storage.get_notifications(client, **filters))
 d.addCallback(
 )
 )
 return d
 def _delete_notification(
-self,
+self, id_: str, is_global: bool, profile_key: str
-id_: str,
-is_global: bool,
-profile_key: str
 ) -> defer.Deferred:
 client = self.host.get_client(profile_key)
 if is_global and not client.is_admin:
 raise exceptions.PermissionError(
 "Only admins can delete global notifications"
 )
 return defer.ensureDeferred(self.delete_notification(client, id_, is_global))
 async def delete_notification(
-self,
+self, client: SatXMPPEntity, id_: str, is_global: bool = False
-client: SatXMPPEntity,
-id_: str,
-is_global: bool=False
 ) -> None:
 """Delete a notification
 the notification must be from the requesting profile.
 @param id_: ID of the notification
 is_global: if True, a global notification will be removed.
 """
 await self.storage.delete_notification(None if is_global else client, id_)
 self.host.bridge.notification_deleted(
-id_,
+id_, C.PROF_KEY_ALL if is_global else client.profile
-C.PROF_KEY_ALL if is_global else client.profile
 )
 def _notifications_expired_clean(
 self, limit_timestamp: float, profile_key: str
 ) -> defer.Deferred:
 else:
 client = self.host.get_client(profile_key)
 return defer.ensureDeferred(
 self.storage.clean_expired_notifications(
-client,
+client, None if limit_timestamp == -1.0 else limit_timestamp
-None if limit_timestamp == -1.0 else limit_timestamp
+)
 )
-)
 ## Misc ##
 def is_entity_available(self, client, entity_jid):
 """Tell from the presence information if the given entity is available.

Mercurial > libervia-backend

comparison libervia/backend/memory/memory.py @ 4270:0d7bb4df2343