Fixed function signatures to uniformly use Optional[] as type hint where appropriate

Author: glpatcern

cs3client/app.py

@@ -7,6 +7,8 @@
 """
 
 import logging
+from typing import Optional
+
 import cs3.app.registry.v1beta1.registry_api_pb2 as cs3arreg
 import cs3.app.registry.v1beta1.resources_pb2 as cs3arres
 import cs3.gateway.v1beta1.gateway_api_pb2 as cs3gw
@@ -44,7 +46,7 @@ def __init__(
         self._config: Config = config
 
     def open_in_app(
-            self, auth_token: tuple, resource: Resource, view_mode: str = None, app: str = None
+            self, auth_token: tuple, resource: Resource, view_mode: Optional[str] = None, app: Optional[str] = None
     ) -> cs3apr.OpenInAppURL:
         """
         Open a file in an app, given the resource, view mode (VIEW_MODE_VIEW_ONLY, VIEW_MODE_READ_ONLY,

cs3client/checkpoint.py

@@ -6,8 +6,9 @@
 Last updated: 30/08/2024
 """
 
-from typing import Generator
+from typing import Generator, Optional
 import logging
+
 import cs3.storage.provider.v1beta1.resources_pb2 as cs3spr
 import cs3.storage.provider.v1beta1.provider_api_pb2 as cs3spp
 from cs3.gateway.v1beta1.gateway_api_pb2_grpc import GatewayAPIStub
@@ -65,7 +66,7 @@ def list_file_versions(
         return res.versions
 
     def restore_file_version(
-            self, auth_token: tuple, resource: Resource, version_key: str, lock_id: str = None
+            self, auth_token: tuple, resource: Resource, version_key: str, lock_id: Optional[str] = None
     ) -> None:
         """
         Restore a file to a previous version.

cs3client/file.py

@@ -10,8 +10,7 @@
 import logging
 import http
 import requests
-from typing import Union
-from typing import Generator
+from typing import Union, Optional, Generator
 import cs3.storage.provider.v1beta1.resources_pb2 as cs3spr
 import cs3.storage.provider.v1beta1.provider_api_pb2 as cs3sp
 from cs3.gateway.v1beta1.gateway_api_pb2_grpc import GatewayAPIStub
@@ -71,7 +70,7 @@ def stat(self, auth_token: tuple, resource: Resource) -> cs3spr.ResourceInfo:
         )
         return res.info
 
-    def set_xattr(self, auth_token: tuple, resource: Resource, key: str, value: str, lock_id: str = None) -> None:
+    def set_xattr(self, auth_token: tuple, resource: Resource, key: str, value: str, lock_id: Optional[str] = None) -> None:
         """
         Set the extended attribute <key> to <value> for a resource.
 
@@ -94,7 +93,7 @@ def set_xattr(self, auth_token: tuple, resource: Resource, key: str, value: str,
         self._status_code_handler.handle_errors(res.status, "set extended attribute", resource.get_file_ref_str())
         self._log.debug(f'msg="Invoked setxattr" trace="{res.status.trace}"')
 
-    def remove_xattr(self, auth_token: tuple, resource: Resource, key: str, lock_id: str = None) -> None:
+    def remove_xattr(self, auth_token: tuple, resource: Resource, key: str, lock_id: Optional[str] = None) -> None:
         """
         Remove the extended attribute <key>.
 
@@ -113,7 +112,7 @@ def remove_xattr(self, auth_token: tuple, resource: Resource, key: str, lock_id:
         self._log.debug(f'msg="Invoked UnsetArbitraryMetaData" trace="{res.status.trace}"')
 
     def rename_file(
-            self, auth_token: tuple, resource: Resource, newresource: Resource, lock_id: str = None
+            self, auth_token: tuple, resource: Resource, newresource: Resource, lock_id: Optional[str] = None
     ) -> None:
         """
         Rename/move resource to new resource.
@@ -133,7 +132,7 @@ def rename_file(
         self._status_code_handler.handle_errors(res.status, "rename file", resource.get_file_ref_str())
         self._log.debug(f'msg="Invoked Move" trace="{res.status.trace}"')
 
-    def remove_file(self, auth_token: tuple, resource: Resource, lock_id: str = None) -> None:
+    def remove_file(self, auth_token: tuple, resource: Resource, lock_id: Optional[str] = None) -> None:
         """
         Remove a resource.
 
@@ -171,7 +170,7 @@ def touch_file(self, auth_token: tuple, resource: Resource) -> None:
 
     def write_file(
             self, auth_token: tuple, resource: Resource, content: Union[str, bytes], size: int,
-            app_name: str = None, lock_id: str = None
+            app_name: Optional[str] = None, lock_id: Optional[str] = None
     ) -> None:
         """
         Write a file using the given userid as access token. The entire content is written
@@ -270,7 +269,7 @@ def write_file(
             f'elapsedTimems="{(tend - tstart) * 1000:.1f}"'
         )
 
-    def read_file(self, auth_token: tuple, resource: Resource, lock_id: str = None) -> Generator[bytes, None, None]:
+    def read_file(self, auth_token: tuple, resource: Resource, lock_id: Optional[str] = None) -> Generator[bytes, None, None]:
         """
         Read a file. Note that the function is a generator, managed by the app server.
 

cs3client/share.py

@@ -7,6 +7,8 @@
 """
 
 import logging
+from typing import Optional
+
 import cs3.sharing.collaboration.v1beta1.collaboration_api_pb2 as cs3scapi
 from cs3.gateway.v1beta1.gateway_api_pb2_grpc import GatewayAPIStub
 import cs3.sharing.collaboration.v1beta1.resources_pb2 as cs3scr
@@ -87,7 +89,7 @@ def create_share(
         return res.share
 
     def list_existing_shares(
-        self, auth_token: tuple, filter_list: list[cs3scr.Filter] = None, page_size: int = 0, page_token: str = None
+        self, auth_token: tuple, filter_list: list[cs3scr.Filter] = None, page_size: int = 0, page_token: Optional[str] = None
     ) -> list[cs3scr.Share]:
         """
         List shares based on a filter.
@@ -109,7 +111,7 @@ def list_existing_shares(
         )
         return (res.share_infos, res.next_page_token)
 
-    def get_share(self, auth_token: tuple, opaque_id: str = None, share_key: cs3scr.ShareKey = None) -> cs3scr.Share:
+    def get_share(self, auth_token: tuple, opaque_id: Optional[str] = None, share_key: Optional[cs3scr.ShareKey] = None) -> cs3scr.Share:
         """
         Get a share by its opaque id or share key (combination of resource_id, grantee and owner),
         one of them is required.
@@ -141,7 +143,7 @@ def get_share(self, auth_token: tuple, opaque_id: str = None, share_key: cs3scr.
         )
         return res.share
 
-    def remove_share(self, auth_token: tuple, opaque_id: str = None, share_key: cs3scr.ShareKey = None) -> None:
+    def remove_share(self, auth_token: tuple, opaque_id: Optional[str] = None, share_key: Optional[cs3scr.ShareKey] = None) -> None:
         """
         Remove a share by its opaque id or share key (combination of resource_id, grantee and owner),
         one of them is required.
@@ -175,9 +177,9 @@ def remove_share(self, auth_token: tuple, opaque_id: str = None, share_key: cs3s
     def update_share(
         self, auth_token: tuple,
         role: str,
-        opaque_id: str = None,
-        share_key: cs3scr.ShareKey = None,
-        display_name: str = None
+        opaque_id: Optional[str] = None,
+        share_key: Optional[cs3scr.ShareKey] = None,
+        display_name: Optional[str] = None
     ) -> cs3scr.Share:
         """
         Update a share by its opaque id.
@@ -216,7 +218,7 @@ def update_share(
         return res.share
 
     def list_received_existing_shares(
-        self, auth_token: tuple, filter_list: list = None, page_size: int = 0, page_token: str = None
+        self, auth_token: tuple, filter_list: Optional[list] = None, page_size: int = 0, page_token: Optional[str] = None
     ) -> list:
         """
         List received existing shares.
@@ -240,7 +242,7 @@ def list_received_existing_shares(
         return (res.share_infos, res.next_page_token)
 
     def get_received_share(
-            self, auth_token: tuple, opaque_id: str = None, share_key: cs3scr.ShareKey = None
+            self, auth_token: tuple, opaque_id: Optional[str] = None, share_key: Optional[cs3scr.ShareKey] = None
     ) -> cs3scr.ReceivedShare:
         """
         Get a received share by its opaque id or share key (combination of resource_id, grantee and owner),
@@ -313,12 +315,12 @@ def create_public_share(
         auth_token: tuple,
         resource_info: cs3spr.ResourceInfo,
         role: str,
-        password: str = None,
-        expiration: cs3types.Timestamp = None,
-        description: str = None,
+        password: Optional[str] = None,
+        expiration: Optional[cs3types.Timestamp] = None,
+        description: Optional[str] = None,
         internal: bool = False,
         notify_uploads: bool = False,
-        notify_uploads_extra_recipients: list = None,
+        notify_uploads_extra_recipients: Optional[list] = None,
     ) -> cs3slr.PublicShare:
         """
         Create a public share.
@@ -358,7 +360,7 @@ def create_public_share(
         return res.share
 
     def list_existing_public_shares(
-        self, auth_token: tuple, filter_list: list = None, page_size: int = 0, page_token: str = None, sign: bool = None
+        self, auth_token: tuple, filter_list: Optional[list] = None, page_size: int = 0, page_token: Optional[str] = None, sign: bool = False
     ) -> list:
         """
         List existing public shares.
@@ -385,7 +387,7 @@ def list_existing_public_shares(
         return (res.share_infos, res.next_page_token)
 
     def get_public_share(
-            self, auth_token: tuple, opaque_id: str = None, token: str = None, sign: bool = False
+            self, auth_token: tuple, opaque_id: Optional[str] = None, token: Optional[str] = None, sign: bool = False
     ) -> cs3slr.PublicShare:
         """
         Get a public share by its opaque id or token, one of them is required.
@@ -423,14 +425,14 @@ def update_public_share(
         auth_token: tuple,
         type: str,
         role: str,
-        opaque_id: str = None,
-        token: str = None,
-        password: str = None,
-        expiration: cs3types.Timestamp = None,
-        notify_uploads_extra_recipients: str = None,
-        description: str = None,
-        display_name: str = None,
-        notify_uploads: bool = None,
+        opaque_id: Optional[str] = None,
+        token: Optional[str] = None,
+        password: Optional[str] = None,
+        expiration: Optional[cs3types.Timestamp] = None,
+        notify_uploads_extra_recipients: Optional[str] = None,
+        description: Optional[str] = None,
+        display_name: Optional[str] = None,
+        notify_uploads: bool = False,
     ) -> None:
         """
         Update a public share by its opaque id or token. (one of them is required), role and type are required,
@@ -486,7 +488,7 @@ def update_public_share(
         )
         return res.share
 
-    def remove_public_share(self, auth_token: tuple, token: str = None, opaque_id: str = None) -> None:
+    def remove_public_share(self, auth_token: tuple, token: Optional[str] = None, opaque_id: Optional[str] = None) -> None:
         """
         Remove a public share by its token or opaque id, one of them is required.
 
@@ -520,12 +522,12 @@ def _create_public_share_update(
         cls,
         type: str,
         role: str,
-        password: str = None,
-        expiration: cs3types.Timestamp = None,
-        display_name: str = None,
-        description: str = None,
-        notify_uploads: bool = None,
-        notify_uploads_extra_recipients: str = None,
+        password: Optional[str] = None,
+        expiration: Optional[cs3types.Timestamp] = None,
+        display_name: Optional[str] = None,
+        description: Optional[str] = None,
+        notify_uploads: bool = False,
+        notify_uploads_extra_recipients: Optional[str] = None,
     ) -> cs3slr.PublicShare:
         """
         Create a public share update object, based on the type the property will be updated,
@@ -567,7 +569,7 @@ def _create_public_share_update(
 
     @classmethod
     def _create_share_grant(
-        cls, opaque_id: str, idp: str, role: str, grantee_type: str, expiration: cs3types.Timestamp = None
+        cls, opaque_id: str, idp: str, role: str, grantee_type: str, expiration: Optional[cs3types.Timestamp] = None
     ) -> cs3scr.ShareGrant:
         """
         Create a share grant object.
@@ -643,10 +645,10 @@ def create_public_share_filter(
         cls,
         filter_type: str,
         resource_id: cs3spr.ResourceId = None,
-        owner_idp: str = None,
-        owner_opaque_id: str = None,
-        creator_idp: str = None,
-        creator_opaque_id: str = None,
+        owner_idp: Optional[str] = None,
+        owner_opaque_id: Optional[str] = None,
+        creator_idp: Optional[str] = None,
+        creator_opaque_id: Optional[str] = None,
     ) -> cs3slapi.ListPublicSharesRequest.Filter:
         """
         Create a public share filter object, based on the filter type (can be TYPE_RESOURCE_ID, TYPE_OWNER,
@@ -688,13 +690,13 @@ def create_share_filter(
         cls,
         filter_type: str,
         resource_id: cs3spr.ResourceId = None,
-        owner_idp: str = None,
-        owner_opaque_id: str = None,
-        creator_idp: str = None,
-        creator_opaque_id: str = None,
-        grantee_type: str = None,
-        space_id: str = None,
-        share_state: str = None,
+        owner_idp: Optional[str] = None,
+        owner_opaque_id: Optional[str] = None,
+        creator_idp: Optional[str] = None,
+        creator_opaque_id: Optional[str] = None,
+        grantee_type: Optional[str] = None,
+        space_id: Optional[str] = None,
+        share_state: Optional[str] = None,
     ) -> cs3scr.Filter:
         """
         Create a share filter object, based on the filter type (can be TYPE_RESOURCE_ID, TYPE_OWNER, TYPE_CREATOR,

cs3client/statuscodehandler.py

@@ -7,6 +7,8 @@
 """
 
 import logging
+from typing import Optional
+
 import cs3.rpc.v1beta1.code_pb2 as cs3code
 import cs3.rpc.v1beta1.status_pb2 as cs3status
 
@@ -20,53 +22,53 @@ def __init__(self, log: logging.Logger, config: Config) -> None:
         self._log = log
         self._config = config
 
-    def _log_not_found_info(self, status: cs3status.Status, operation: str, status_msg: str, msg: str = None) -> None:
+    def _log_not_found_info(self, status: cs3status.Status, operation: str, status_msg: str, msg: Optional[str] = None) -> None:
         self._log.info(
             f'msg="Not found on {operation}" {msg + " " if msg else ""} '
             f'userid="{self._config.auth_client_id if self._config.auth_client_id else "no_id_set"}" '
             f'trace="{status.trace}" reason="{status_msg}"'
         )
 
     def _log_authentication_error(
-            self, status: cs3status.Status, operation: str, status_msg: str, msg: str = None
+            self, status: cs3status.Status, operation: str, status_msg: str, msg: Optional[str] = None
     ) -> None:
         self._log.error(
             f'msg="Authentication failed on {operation}" {msg + " " if msg else ""}'
             f'userid="{self._config.auth_client_id if self._config.auth_client_id else "no_id_set"}" '
             f'trace="{status.trace}" reason="{status_msg}"'
         )
 
-    def _log_unknown_error(self, status: cs3status.Status, operation: str, status_msg: str, msg: str = None) -> None:
+    def _log_unknown_error(self, status: cs3status.Status, operation: str, status_msg: str, msg: Optional[str] = None) -> None:
         self._log.error(
             f'msg="Failed to {operation}, unknown error" {msg + " " if msg else ""}'
             f'userid="{self._config.auth_client_id if self._config.auth_client_id else "no_id_set"}" '
             f'trace="{status.trace}" reason="{status_msg}"'
         )
 
     def _log_precondition_info(
-            self, status: cs3status.Status, operation: str, status_msg: str, msg: str = None
+            self, status: cs3status.Status, operation: str, status_msg: str, msg: Optional[str] = None
     ) -> None:
         self._log.info(
             f'msg="Failed precondition on {operation}" {msg + " " if msg else ""}'
             f'userid="{self._config.auth_client_id if self._config.auth_client_id else "no_id_set"}" '
             f'trace="{status.trace}" reason="{status_msg}"'
         )
 
-    def _log_already_exists(self, status: cs3status.Status, operation: str, status_msg: str, msg: str = None) -> None:
+    def _log_already_exists(self, status: cs3status.Status, operation: str, status_msg: str, msg: Optional[str] = None) -> None:
         self._log.info(
             f'msg="Already exists on {operation}" {msg + " " if msg else ""}'
             f'userid="{self._config.auth_client_id if self._config.auth_client_id else "no_id_set"}" '
             f'trace="{status.trace}" reason="{status_msg}"'
         )
 
-    def _log_unimplemented(self, status: cs3status.Status, operation: str, status_msg: str, msg: str = None) -> None:
+    def _log_unimplemented(self, status: cs3status.Status, operation: str, status_msg: str, msg: Optional[str] = None) -> None:
         self._log.info(
             f'msg="Invoked {operation} on unimplemented feature" {msg + " " if msg else ""}'
             f'userid="{self._config.auth_client_id if self._config.auth_client_id else "no_id_set"}" '
             f'trace="{status.trace}" reason="{status_msg}"'
         )
 
-    def handle_errors(self, status: cs3status.Status, operation: str, msg: str = None) -> None:
+    def handle_errors(self, status: cs3status.Status, operation: str, msg: Optional[str] = None) -> None:
 
         if status.code == cs3code.CODE_OK:
             return