cosmetics: cosmetics, abstract a function
This commit is contained in:
@@ -80,7 +80,7 @@ def xml_proppatch(base_prefix: str, path: str,
|
|||||||
all_props = radicale_item.check_and_sanitize_props(all_props_with_remove)
|
all_props = radicale_item.check_and_sanitize_props(all_props_with_remove)
|
||||||
if share and share_overlay and _sharing is not None:
|
if share and share_overlay and _sharing is not None:
|
||||||
# _sharing is not None: always the case, but makes mypy happy
|
# _sharing is not None: always the case, but makes mypy happy
|
||||||
_sharing.database_update_sharing(ShareType=share['ShareType'],
|
_sharing.sharing_collection_update(ShareType=share['ShareType'],
|
||||||
PathOrToken=share['PathOrToken'],
|
PathOrToken=share['PathOrToken'],
|
||||||
OwnerOrUser=share['User'],
|
OwnerOrUser=share['User'],
|
||||||
Properties=cast(Dict[str, str], all_props))
|
Properties=cast(Dict[str, str], all_props))
|
||||||
|
|||||||
@@ -162,7 +162,7 @@ class BaseSharing:
|
|||||||
else:
|
else:
|
||||||
logger.info("sharing database info: (not provided)")
|
logger.info("sharing database info: (not provided)")
|
||||||
|
|
||||||
# overloadable database functions
|
# *** overloadable database functions ***
|
||||||
def database_init(self) -> bool:
|
def database_init(self) -> bool:
|
||||||
""" initialize db """
|
""" initialize db """
|
||||||
return False
|
return False
|
||||||
@@ -230,7 +230,7 @@ class BaseSharing:
|
|||||||
""" delete sharing """
|
""" delete sharing """
|
||||||
return {"status": "not-implemented"}
|
return {"status": "not-implemented"}
|
||||||
|
|
||||||
# sharing functions called by request methods
|
# *** functions called by cli ***
|
||||||
def verify(self) -> bool:
|
def verify(self) -> bool:
|
||||||
""" verify database """
|
""" verify database """
|
||||||
logger.info("sharing database verification begin")
|
logger.info("sharing database verification begin")
|
||||||
@@ -263,6 +263,29 @@ class BaseSharing:
|
|||||||
logger.info("sharing database verification content successful")
|
logger.info("sharing database verification content successful")
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
# *** sharing functions called by request methods ***
|
||||||
|
# list sharings of type "map"
|
||||||
|
def sharing_collection_map_list(self, User: Union[str, None] = None, Enabled: Union[bool, None] = None, Hidden: Union[bool, None] = None) -> list[dict]:
|
||||||
|
""" returning dict with shared collections by filter(User/Enabled/Hidden) or None if not found"""
|
||||||
|
if not self.sharing_collection_by_map:
|
||||||
|
if logger.isEnabledFor(logging.DEBUG):
|
||||||
|
logger.debug("TRACE/sharing/map: not active")
|
||||||
|
return [{}]
|
||||||
|
|
||||||
|
# retrieve collections depending on filter
|
||||||
|
shared_collection_list = self.database_list_sharing(
|
||||||
|
ShareType="map",
|
||||||
|
OwnerOrUser=User,
|
||||||
|
User=User,
|
||||||
|
EnabledByOwner=Enabled,
|
||||||
|
EnabledByUser=Enabled,
|
||||||
|
HiddenByOwner=Hidden,
|
||||||
|
HiddenByUser=Hidden)
|
||||||
|
|
||||||
|
# final
|
||||||
|
return shared_collection_list
|
||||||
|
|
||||||
|
# resolves a path to a share
|
||||||
def sharing_collection_resolver(self, path: str, user: str) -> Union[dict, None]:
|
def sharing_collection_resolver(self, path: str, user: str) -> Union[dict, None]:
|
||||||
""" returning dict with PathMapped, Owner, Permissions or None if not found"""
|
""" returning dict with PathMapped, Owner, Permissions or None if not found"""
|
||||||
if self.sharing_collection_by_token:
|
if self.sharing_collection_by_token:
|
||||||
@@ -286,31 +309,19 @@ class BaseSharing:
|
|||||||
logger.debug("TRACE/sharing/map: not active")
|
logger.debug("TRACE/sharing/map: not active")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# final
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# list sharings of type "map"
|
# adjust a share
|
||||||
def sharing_collection_map_list(self, User: Union[str, None] = None, Enabled: Union[bool, None] = None, Hidden: Union[bool, None] = None) -> list[dict]:
|
def sharing_collection_update(self, ShareType: str, PathOrToken: str, OwnerOrUser: str, Properties: dict) -> None:
|
||||||
""" returning dict with shared collections by filter(User/Enabled/Hidden) or None if not found"""
|
""" returning dict with PathMapped, Owner, Permissions or None if not found"""
|
||||||
if not self.sharing_collection_by_map:
|
logger.info("Sharing/collection/update: ShareType=%r PathOrToken=%r OwnerOrUser=%r", ShareType, PathOrToken, OwnerOrUser)
|
||||||
if logger.isEnabledFor(logging.DEBUG):
|
self.database_update_sharing(ShareType=ShareType,
|
||||||
logger.debug("TRACE/sharing/map: not active")
|
PathOrToken=PathOrToken,
|
||||||
return [{}]
|
OwnerOrUser=OwnerOrUser,
|
||||||
|
Properties=Properties)
|
||||||
|
|
||||||
# retrieve collections depending on filter
|
# *** internal sharing functions ***
|
||||||
shared_collection_list = self.database_list_sharing(
|
# resolves a token "path" to a share
|
||||||
ShareType="map",
|
|
||||||
OwnerOrUser=User,
|
|
||||||
User=User,
|
|
||||||
EnabledByOwner=Enabled,
|
|
||||||
EnabledByUser=Enabled,
|
|
||||||
HiddenByOwner=Hidden,
|
|
||||||
HiddenByUser=Hidden)
|
|
||||||
|
|
||||||
# final
|
|
||||||
return shared_collection_list
|
|
||||||
|
|
||||||
# internal sharing functions
|
|
||||||
def sharing_collection_by_token_resolver(self, path) -> Union[dict, None]:
|
def sharing_collection_by_token_resolver(self, path) -> Union[dict, None]:
|
||||||
""" returning dict with PathMapped, Owner, Permissions or None if invalid"""
|
""" returning dict with PathMapped, Owner, Permissions or None if invalid"""
|
||||||
if self.sharing_collection_by_token:
|
if self.sharing_collection_by_token:
|
||||||
@@ -339,6 +350,7 @@ class BaseSharing:
|
|||||||
logger.debug("TRACE/sharing/token: not active")
|
logger.debug("TRACE/sharing/token: not active")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
# resolves a map "path" to a share
|
||||||
def sharing_collection_by_map_resolver(self, path: str, user: str) -> Union[dict, None]:
|
def sharing_collection_by_map_resolver(self, path: str, user: str) -> Union[dict, None]:
|
||||||
""" returning dict with PathMapped, Owner, Permissions or None if invalid"""
|
""" returning dict with PathMapped, Owner, Permissions or None if invalid"""
|
||||||
if self.sharing_collection_by_map:
|
if self.sharing_collection_by_map:
|
||||||
@@ -373,7 +385,7 @@ class BaseSharing:
|
|||||||
logger.debug("TRACE/sharing/map: not active")
|
logger.debug("TRACE/sharing/map: not active")
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# POST API
|
# *** POST API ***
|
||||||
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str) -> types.WSGIResponse:
|
def post(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str) -> types.WSGIResponse:
|
||||||
# Late import to avoid circular dependency in config
|
# Late import to avoid circular dependency in config
|
||||||
from radicale.app.base import Access
|
from radicale.app.base import Access
|
||||||
|
|||||||
Reference in New Issue
Block a user