diff --git a/CHANGELOG.md b/CHANGELOG.md index 02092849..39479aa7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,12 +2,17 @@ All notable changes to this project will be documented in this file. -## [0.6.1 - 202x-xx-xx] +## [0.7.0 - 2022-12-2x] ### Added - set_handlers: `enabled_handler`, `heartbeat_handler` now can be async(Coroutines). #175 +### Changed + +- drop Python 3.9 support. #180 +- internal code refactoring and clean-up #177 + ## [0.6.0 - 2023-12-06] ### Added diff --git a/README.md b/README.md index 02ca9be5..f0abc8f5 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ [![codecov](https://codecov.io/github/cloud-py-api/nc_py_api/branch/main/graph/badge.svg?token=C91PL3FYDQ)](https://codecov.io/github/cloud-py-api/nc_py_api) ![NextcloudVersion](https://img.shields.io/badge/Nextcloud-26%20%7C%2027%20%7C%2028-blue) -![PythonVersion](https://img.shields.io/badge/python-3.9%20%7C%203.10%20%7C%203.11%20%7C%203.12-blue) +![PythonVersion](https://img.shields.io/badge/python-3.10%20%7C%203.11%20%7C%203.12-blue) ![impl](https://img.shields.io/pypi/implementation/nc_py_api) ![pypi](https://img.shields.io/pypi/v/nc_py_api.svg) diff --git a/benchmarks/aa_overhead_dav_download.py b/benchmarks/aa_overhead_dav_download.py index a923e0e1..8161546c 100644 --- a/benchmarks/aa_overhead_dav_download.py +++ b/benchmarks/aa_overhead_dav_download.py @@ -1,7 +1,7 @@ from getpass import getuser from random import randbytes from time import perf_counter -from typing import Any, Union +from typing import Any import matplotlib.pyplot as plt from aa_overhead_common import measure_overhead, os_id @@ -12,7 +12,7 @@ CACHE_SESS = False -def measure_download_1mb(nc_obj: Union[Nextcloud, NextcloudApp]) -> [Any, float]: +def measure_download_1mb(nc_obj: Nextcloud | NextcloudApp) -> [Any, float]: __result = None small_file_name = "1Mb.bin" small_file = randbytes(1024 * 1024) diff --git a/benchmarks/aa_overhead_dav_download_stream.py b/benchmarks/aa_overhead_dav_download_stream.py index 899a5070..07a9b232 100644 --- a/benchmarks/aa_overhead_dav_download_stream.py +++ b/benchmarks/aa_overhead_dav_download_stream.py @@ -2,7 +2,7 @@ from io import BytesIO from random import randbytes from time import perf_counter -from typing import Any, Union +from typing import Any import matplotlib.pyplot as plt from aa_overhead_common import measure_overhead, os_id @@ -13,7 +13,7 @@ CACHE_SESS = False -def measure_download_100mb(nc_obj: Union[Nextcloud, NextcloudApp]) -> [Any, float]: +def measure_download_100mb(nc_obj: Nextcloud | NextcloudApp) -> [Any, float]: __result = None medium_file_name = "100Mb.bin" medium_file = BytesIO() diff --git a/benchmarks/aa_overhead_dav_upload.py b/benchmarks/aa_overhead_dav_upload.py index 7268b529..3661ed74 100644 --- a/benchmarks/aa_overhead_dav_upload.py +++ b/benchmarks/aa_overhead_dav_upload.py @@ -1,7 +1,7 @@ from getpass import getuser from random import randbytes from time import perf_counter -from typing import Any, Union +from typing import Any import matplotlib.pyplot as plt from aa_overhead_common import measure_overhead, os_id @@ -12,7 +12,7 @@ CACHE_SESS = False -def measure_upload_1mb(nc_obj: Union[Nextcloud, NextcloudApp]) -> [Any, float]: +def measure_upload_1mb(nc_obj: Nextcloud | NextcloudApp) -> [Any, float]: __result = None small_file_name = "1Mb.bin" small_file = randbytes(1024 * 1024) diff --git a/benchmarks/aa_overhead_dav_upload_stream.py b/benchmarks/aa_overhead_dav_upload_stream.py index 481aecbd..1bd150cc 100644 --- a/benchmarks/aa_overhead_dav_upload_stream.py +++ b/benchmarks/aa_overhead_dav_upload_stream.py @@ -2,7 +2,7 @@ from io import BytesIO from random import randbytes from time import perf_counter -from typing import Any, Union +from typing import Any import matplotlib.pyplot as plt from aa_overhead_common import measure_overhead, os_id @@ -13,7 +13,7 @@ CACHE_SESS = False -def measure_upload_100mb(nc_obj: Union[Nextcloud, NextcloudApp]) -> [Any, float]: +def measure_upload_100mb(nc_obj: Nextcloud | NextcloudApp) -> [Any, float]: __result = None medium_file_name = "100Mb.bin" medium_file = BytesIO() diff --git a/benchmarks/aa_overhead_ocs.py b/benchmarks/aa_overhead_ocs.py index 78039ea2..40550588 100644 --- a/benchmarks/aa_overhead_ocs.py +++ b/benchmarks/aa_overhead_ocs.py @@ -1,6 +1,6 @@ from getpass import getuser from time import perf_counter -from typing import Any, Union +from typing import Any import matplotlib.pyplot as plt from aa_overhead_common import measure_overhead, os_id @@ -11,7 +11,7 @@ CACHE_SESS = False -def measure_get_details(nc_obj: Union[Nextcloud, NextcloudApp]) -> [Any, float]: +def measure_get_details(nc_obj: Nextcloud | NextcloudApp) -> [Any, float]: __result = None start_time = perf_counter() for _ in range(ITERS): diff --git a/benchmarks/conf.py b/benchmarks/conf.py index d0f12d0c..766a97d8 100644 --- a/benchmarks/conf.py +++ b/benchmarks/conf.py @@ -1,5 +1,3 @@ -from typing import Optional - from nc_py_api import Nextcloud, NextcloudApp NC_CFGS = { @@ -39,19 +37,19 @@ } -def init_nc(url, cfg) -> Optional[Nextcloud]: +def init_nc(url, cfg) -> Nextcloud | None: if cfg.get("nc_auth_user", "") and cfg.get("nc_auth_pass", ""): return Nextcloud(nc_auth_user=cfg["nc_auth_user"], nc_auth_pass=cfg["nc_auth_pass"], nextcloud_url=url) return None -def init_nc_by_app_pass(url, cfg) -> Optional[Nextcloud]: +def init_nc_by_app_pass(url, cfg) -> Nextcloud | None: if cfg.get("nc_auth_user", "") and cfg.get("nc_auth_app_pass", ""): return Nextcloud(nc_auth_user=cfg["nc_auth_user"], nc_auth_pass=cfg["nc_auth_app_pass"], nextcloud_url=url) return None -def init_nc_app(url, cfg) -> Optional[NextcloudApp]: +def init_nc_app(url, cfg) -> NextcloudApp | None: if cfg.get("secret", "") and cfg.get("app_id", ""): return NextcloudApp( app_id=cfg["app_id"], diff --git a/nc_py_api/_misc.py b/nc_py_api/_misc.py index 2ca5ab73..a7a01438 100644 --- a/nc_py_api/_misc.py +++ b/nc_py_api/_misc.py @@ -5,9 +5,9 @@ import secrets from base64 import b64decode +from collections.abc import Callable from datetime import datetime, timezone from string import ascii_letters, digits -from typing import Callable, Union from ._exceptions import NextcloudMissingCapabilities @@ -28,7 +28,7 @@ def clear_from_params_empty(keys: list[str], params: dict) -> None: params.pop(key) -def require_capabilities(capabilities: Union[str, list[str]], srv_capabilities: dict) -> None: +def require_capabilities(capabilities: str | list[str], srv_capabilities: dict) -> None: """Checks for capabilities and raises an exception if any of them are missing.""" result = check_capabilities(capabilities, srv_capabilities) if result: @@ -52,7 +52,7 @@ def __check_sub_capability(split_capabilities: list[str], srv_capabilities: dict return True -def check_capabilities(capabilities: Union[str, list[str]], srv_capabilities: dict) -> list[str]: +def check_capabilities(capabilities: str | list[str], srv_capabilities: dict) -> list[str]: """Checks for capabilities and returns a list of missing ones.""" if isinstance(capabilities, str): capabilities = [capabilities] diff --git a/nc_py_api/_preferences_ex.py b/nc_py_api/_preferences_ex.py index f5d0c72d..f9682369 100644 --- a/nc_py_api/_preferences_ex.py +++ b/nc_py_api/_preferences_ex.py @@ -1,7 +1,6 @@ """Nextcloud API for working with apps V2's storage w/wo user context(table oc_appconfig_ex/oc_preferences_ex).""" import dataclasses -import typing from ._exceptions import NextcloudExceptionNotFound from ._misc import require_capabilities @@ -26,7 +25,7 @@ class _BasicAppCfgPref: def __init__(self, session: NcSessionBasic): self._session = session - def get_value(self, key: str, default=None) -> typing.Optional[str]: + def get_value(self, key: str, default=None) -> str | None: """Returns the value of the key, if found, or the specified default value.""" if not key: raise ValueError("`key` parameter can not be empty") @@ -47,7 +46,7 @@ def get_values(self, keys: list[str]) -> list[CfgRecord]: results = self._session.ocs("POST", f"{self._session.ae_url}/{self._url_suffix}/get-values", json=data) return [CfgRecord(i) for i in results] - def delete(self, keys: typing.Union[str, list[str]], not_fail=True) -> None: + def delete(self, keys: str | list[str], not_fail=True) -> None: """Deletes config/preference entries by the provided keys.""" if isinstance(keys, str): keys = [keys] @@ -82,7 +81,7 @@ class AppConfigExAPI(_BasicAppCfgPref): _url_suffix = "ex-app/config" - def set_value(self, key: str, value: str, sensitive: typing.Optional[bool] = None) -> None: + def set_value(self, key: str, value: str, sensitive: bool | None = None) -> None: """Sets a value and if specified the sensitive flag for a key. .. note:: A sensitive flag ensures key values are truncated in Nextcloud logs. diff --git a/nc_py_api/_session.py b/nc_py_api/_session.py index d85e0be4..3ef78490 100644 --- a/nc_py_api/_session.py +++ b/nc_py_api/_session.py @@ -49,9 +49,9 @@ class ServerVersion(typing.TypedDict): @dataclass class RuntimeOptions: xdebug_session: str - timeout: typing.Optional[int] - timeout_dav: typing.Optional[int] - _nc_cert: typing.Union[str, bool] + timeout: int | None + timeout_dav: int | None + _nc_cert: str | bool upload_chunk_v2: bool def __init__(self, **kwargs): @@ -62,7 +62,7 @@ def __init__(self, **kwargs): self.upload_chunk_v2 = kwargs.get("chunked_upload_v2", options.CHUNKED_UPLOAD_V2) @property - def nc_cert(self) -> typing.Union[str, bool]: + def nc_cert(self) -> str | bool: return self._nc_cert @@ -158,9 +158,9 @@ def ocs( method: str, path: str, *, - content: typing.Optional[typing.Union[bytes, str, typing.Iterable[bytes], typing.AsyncIterable[bytes]]] = None, - json: typing.Optional[typing.Union[dict, list]] = None, - params: typing.Optional[dict] = None, + content: bytes | str | typing.Iterable[bytes] | typing.AsyncIterable[bytes] | None = None, + json: dict | list | None = None, + params: dict | None = None, **kwargs, ): self.init_adapter() diff --git a/nc_py_api/_talk_api.py b/nc_py_api/_talk_api.py index c8830e7e..3311cc2a 100644 --- a/nc_py_api/_talk_api.py +++ b/nc_py_api/_talk_api.py @@ -1,7 +1,6 @@ """Nextcloud Talk API implementation.""" import hashlib -import typing from ._exceptions import check_error from ._misc import ( @@ -51,7 +50,7 @@ def bots_available(self) -> bool: return not check_capabilities("spreed.features.bots-v1", self._session.capabilities) def get_user_conversations( - self, no_status_update: bool = True, include_status: bool = False, modified_since: typing.Union[int, bool] = 0 + self, no_status_update: bool = True, include_status: bool = False, modified_since: int | bool = 0 ) -> list[Conversation]: """Returns the list of the user's conversations. @@ -78,9 +77,7 @@ def get_user_conversations( self._update_config_sha() return [Conversation(i) for i in result] - def list_participants( - self, conversation: typing.Union[Conversation, str], include_status: bool = False - ) -> list[Participant]: + def list_participants(self, conversation: Conversation | str, include_status: bool = False) -> list[Participant]: """Returns a list of conversation participants. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -127,7 +124,7 @@ def create_conversation( clear_from_params_empty(["invite", "source", "roomName", "objectType", "objectId"], params) return Conversation(self._session.ocs("POST", self._ep_base + "/api/v4/room", json=params)) - def rename_conversation(self, conversation: typing.Union[Conversation, str], new_name: str) -> None: + def rename_conversation(self, conversation: Conversation | str, new_name: str) -> None: """Renames a conversation. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -136,7 +133,7 @@ def rename_conversation(self, conversation: typing.Union[Conversation, str], new token = conversation.token if isinstance(conversation, Conversation) else conversation self._session.ocs("PUT", self._ep_base + f"/api/v4/room/{token}", params={"roomName": new_name}) - def set_conversation_description(self, conversation: typing.Union[Conversation, str], description: str) -> None: + def set_conversation_description(self, conversation: Conversation | str, description: str) -> None: """Sets conversation description. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -147,7 +144,7 @@ def set_conversation_description(self, conversation: typing.Union[Conversation, "PUT", self._ep_base + f"/api/v4/room/{token}/description", params={"description": description} ) - def set_conversation_fav(self, conversation: typing.Union[Conversation, str], favorite: bool) -> None: + def set_conversation_fav(self, conversation: Conversation | str, favorite: bool) -> None: """Changes conversation **favorite** state. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -156,7 +153,7 @@ def set_conversation_fav(self, conversation: typing.Union[Conversation, str], fa token = conversation.token if isinstance(conversation, Conversation) else conversation self._session.ocs("POST" if favorite else "DELETE", self._ep_base + f"/api/v4/room/{token}/favorite") - def set_conversation_password(self, conversation: typing.Union[Conversation, str], password: str) -> None: + def set_conversation_password(self, conversation: Conversation | str, password: str) -> None: """Sets password for a conversation. Currently, it is only allowed to have a password for ``public`` conversations. @@ -169,7 +166,7 @@ def set_conversation_password(self, conversation: typing.Union[Conversation, str token = conversation.token if isinstance(conversation, Conversation) else conversation self._session.ocs("PUT", self._ep_base + f"/api/v4/room/{token}/password", params={"password": password}) - def set_conversation_readonly(self, conversation: typing.Union[Conversation, str], read_only: bool) -> None: + def set_conversation_readonly(self, conversation: Conversation | str, read_only: bool) -> None: """Changes conversation **read_only** state. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -178,7 +175,7 @@ def set_conversation_readonly(self, conversation: typing.Union[Conversation, str token = conversation.token if isinstance(conversation, Conversation) else conversation self._session.ocs("PUT", self._ep_base + f"/api/v4/room/{token}/read-only", params={"state": int(read_only)}) - def set_conversation_public(self, conversation: typing.Union[Conversation, str], public: bool) -> None: + def set_conversation_public(self, conversation: Conversation | str, public: bool) -> None: """Changes conversation **public** state. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -187,9 +184,7 @@ def set_conversation_public(self, conversation: typing.Union[Conversation, str], token = conversation.token if isinstance(conversation, Conversation) else conversation self._session.ocs("POST" if public else "DELETE", self._ep_base + f"/api/v4/room/{token}/public") - def set_conversation_notify_lvl( - self, conversation: typing.Union[Conversation, str], new_lvl: NotificationLevel - ) -> None: + def set_conversation_notify_lvl(self, conversation: Conversation | str, new_lvl: NotificationLevel) -> None: """Sets new notification level for user in the conversation. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -198,14 +193,14 @@ def set_conversation_notify_lvl( token = conversation.token if isinstance(conversation, Conversation) else conversation self._session.ocs("POST", self._ep_base + f"/api/v4/room/{token}/notify", json={"level": int(new_lvl)}) - def get_conversation_by_token(self, conversation: typing.Union[Conversation, str]) -> Conversation: + def get_conversation_by_token(self, conversation: Conversation | str) -> Conversation: """Gets conversation by token.""" token = conversation.token if isinstance(conversation, Conversation) else conversation result = self._session.ocs("GET", self._ep_base + f"/api/v4/room/{token}") self._update_config_sha() return Conversation(result) - def delete_conversation(self, conversation: typing.Union[Conversation, str]) -> None: + def delete_conversation(self, conversation: Conversation | str) -> None: """Deletes a conversation. .. note:: Deleting a conversation that is the parent of breakout rooms, will also delete them. @@ -217,7 +212,7 @@ def delete_conversation(self, conversation: typing.Union[Conversation, str]) -> token = conversation.token if isinstance(conversation, Conversation) else conversation self._session.ocs("DELETE", self._ep_base + f"/api/v4/room/{token}") - def leave_conversation(self, conversation: typing.Union[Conversation, str]) -> None: + def leave_conversation(self, conversation: Conversation | str) -> None: """Removes yourself from the conversation. .. note:: When the participant is a moderator or owner and there are no other moderators or owners left, @@ -231,8 +226,8 @@ def leave_conversation(self, conversation: typing.Union[Conversation, str]) -> N def send_message( self, message: str, - conversation: typing.Union[Conversation, str] = "", - reply_to_message: typing.Union[int, TalkMessage] = 0, + conversation: Conversation | str = "", + reply_to_message: int | TalkMessage = 0, silent: bool = False, actor_display_name: str = "", ) -> TalkMessage: @@ -262,11 +257,7 @@ def send_message( r = self._session.ocs("POST", self._ep_base + f"/api/v1/chat/{token}", json=params) return TalkMessage(r) - def send_file( - self, - path: typing.Union[str, FsNode], - conversation: typing.Union[Conversation, str] = "", - ) -> tuple[Share, str]: + def send_file(self, path: str | FsNode, conversation: Conversation | str = "") -> tuple[Share, str]: require_capabilities("files_sharing.api_enabled", self._session.capabilities) token = conversation.token if isinstance(conversation, Conversation) else conversation reference_id = hashlib.sha256(random_string(32).encode("UTF-8")).hexdigest() @@ -281,7 +272,7 @@ def send_file( def receive_messages( self, - conversation: typing.Union[Conversation, str], + conversation: Conversation | str, look_in_future: bool = False, limit: int = 100, timeout: int = 30, @@ -305,9 +296,7 @@ def receive_messages( result = self._session.ocs("GET", self._ep_base + f"/api/v1/chat/{token}", params=params) return [TalkFileMessage(i, self._session.user) if i["message"] == "{file}" else TalkMessage(i) for i in result] - def delete_message( - self, message: typing.Union[TalkMessage, str], conversation: typing.Union[Conversation, str] = "" - ) -> TalkMessage: + def delete_message(self, message: TalkMessage | str, conversation: Conversation | str = "") -> TalkMessage: """Delete a chat message. :param message: Message ID or :py:class:`~nc_py_api.talk.TalkMessage` to delete. @@ -321,10 +310,7 @@ def delete_message( return TalkMessage(result) def react_to_message( - self, - message: typing.Union[TalkMessage, str], - reaction: str, - conversation: typing.Union[Conversation, str] = "", + self, message: TalkMessage | str, reaction: str, conversation: Conversation | str = "" ) -> dict[str, list[MessageReactions]]: """React to a chat message. @@ -345,10 +331,7 @@ def react_to_message( return {k: [MessageReactions(i) for i in v] for k, v in r.items()} if r else {} def delete_reaction( - self, - message: typing.Union[TalkMessage, str], - reaction: str, - conversation: typing.Union[Conversation, str] = "", + self, message: TalkMessage | str, reaction: str, conversation: Conversation | str = "" ) -> dict[str, list[MessageReactions]]: """Remove reaction from a chat message. @@ -367,10 +350,7 @@ def delete_reaction( return {k: [MessageReactions(i) for i in v] for k, v in r.items()} if r else {} def get_message_reactions( - self, - message: typing.Union[TalkMessage, str], - reaction_filter: str = "", - conversation: typing.Union[Conversation, str] = "", + self, message: TalkMessage | str, reaction_filter: str = "", conversation: Conversation | str = "" ) -> dict[str, list[MessageReactions]]: """Get reactions information for a chat message. @@ -391,7 +371,7 @@ def list_bots(self) -> list[BotInfo]: require_capabilities("spreed.features.bots-v1", self._session.capabilities) return [BotInfo(i) for i in self._session.ocs("GET", self._ep_base + "/api/v1/bot/admin")] - def conversation_list_bots(self, conversation: typing.Union[Conversation, str]) -> list[BotInfoBasic]: + def conversation_list_bots(self, conversation: Conversation | str) -> list[BotInfoBasic]: """Lists the bots that are enabled and can be enabled for the conversation. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -400,7 +380,7 @@ def conversation_list_bots(self, conversation: typing.Union[Conversation, str]) token = conversation.token if isinstance(conversation, Conversation) else conversation return [BotInfoBasic(i) for i in self._session.ocs("GET", self._ep_base + f"/api/v1/bot/{token}")] - def enable_bot(self, conversation: typing.Union[Conversation, str], bot: typing.Union[BotInfoBasic, int]) -> None: + def enable_bot(self, conversation: Conversation | str, bot: BotInfoBasic | int) -> None: """Enable a bot for a conversation as a moderator. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -411,7 +391,7 @@ def enable_bot(self, conversation: typing.Union[Conversation, str], bot: typing. bot_id = bot.bot_id if isinstance(bot, BotInfoBasic) else bot self._session.ocs("POST", self._ep_base + f"/api/v1/bot/{token}/{bot_id}") - def disable_bot(self, conversation: typing.Union[Conversation, str], bot: typing.Union[BotInfoBasic, int]) -> None: + def disable_bot(self, conversation: Conversation | str, bot: BotInfoBasic | int) -> None: """Disable a bot for a conversation as a moderator. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -424,7 +404,7 @@ def disable_bot(self, conversation: typing.Union[Conversation, str], bot: typing def create_poll( self, - conversation: typing.Union[Conversation, str], + conversation: Conversation | str, question: str, options: list[str], hidden_results: bool = True, @@ -447,7 +427,7 @@ def create_poll( } return Poll(self._session.ocs("POST", self._ep_base + f"/api/v1/poll/{token}", json=params), token) - def get_poll(self, poll: typing.Union[Poll, int], conversation: typing.Union[Conversation, str] = "") -> Poll: + def get_poll(self, poll: Poll | int, conversation: Conversation | str = "") -> Poll: """Get state or result of a poll. :param poll: Poll ID or :py:class:`~nc_py_api.talk.Poll`. @@ -461,12 +441,7 @@ def get_poll(self, poll: typing.Union[Poll, int], conversation: typing.Union[Con token = conversation.token if isinstance(conversation, Conversation) else conversation return Poll(self._session.ocs("GET", self._ep_base + f"/api/v1/poll/{token}/{poll_id}"), token) - def vote_poll( - self, - options_ids: list[int], - poll: typing.Union[Poll, int], - conversation: typing.Union[Conversation, str] = "", - ) -> Poll: + def vote_poll(self, options_ids: list[int], poll: Poll | int, conversation: Conversation | str = "") -> Poll: """Vote on a poll. :param options_ids: The option IDs the participant wants to vote for. @@ -484,7 +459,7 @@ def vote_poll( ) return Poll(r, token) - def close_poll(self, poll: typing.Union[Poll, int], conversation: typing.Union[Conversation, str] = "") -> Poll: + def close_poll(self, poll: Poll | int, conversation: Conversation | str = "") -> Poll: """Close a poll. :param poll: Poll ID or :py:class:`~nc_py_api.talk.Poll`. @@ -499,9 +474,7 @@ def close_poll(self, poll: typing.Union[Poll, int], conversation: typing.Union[C return Poll(self._session.ocs("DELETE", self._ep_base + f"/api/v1/poll/{token}/{poll_id}"), token) def set_conversation_avatar( - self, - conversation: typing.Union[Conversation, str], - avatar: typing.Union[bytes, tuple[str, typing.Union[str, None]]], + self, conversation: Conversation | str, avatar: bytes | tuple[str, str | None] ) -> Conversation: """Set image or emoji as avatar for the conversation. @@ -526,7 +499,7 @@ def set_conversation_avatar( ) return Conversation(r) - def delete_conversation_avatar(self, conversation: typing.Union[Conversation, str]) -> Conversation: + def delete_conversation_avatar(self, conversation: Conversation | str) -> Conversation: """Delete conversation avatar. :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -535,7 +508,7 @@ def delete_conversation_avatar(self, conversation: typing.Union[Conversation, st token = conversation.token if isinstance(conversation, Conversation) else conversation return Conversation(self._session.ocs("DELETE", self._ep_base + f"/api/v1/room/{token}/avatar")) - def get_conversation_avatar(self, conversation: typing.Union[Conversation, str], dark=False) -> bytes: + def get_conversation_avatar(self, conversation: Conversation | str, dark=False) -> bytes: """Get conversation avatar (binary). :param conversation: conversation token or :py:class:`~nc_py_api.talk.Conversation`. @@ -549,7 +522,7 @@ def get_conversation_avatar(self, conversation: typing.Union[Conversation, str], return response.content @staticmethod - def _get_token(message: typing.Union[TalkMessage, str], conversation: typing.Union[Conversation, str]) -> str: + def _get_token(message: TalkMessage | str, conversation: Conversation | str) -> str: if not conversation and not isinstance(message, TalkMessage): raise ValueError("Either specify 'conversation' or provide 'TalkMessage'.") diff --git a/nc_py_api/_version.py b/nc_py_api/_version.py index 635b0804..a2c87cfe 100644 --- a/nc_py_api/_version.py +++ b/nc_py_api/_version.py @@ -1,3 +1,3 @@ """Version of nc_py_api.""" -__version__ = "0.6.0" +__version__ = "0.7.0.dev0" diff --git a/nc_py_api/activity.py b/nc_py_api/activity.py index 0f646db6..f4dc6798 100644 --- a/nc_py_api/activity.py +++ b/nc_py_api/activity.py @@ -2,7 +2,6 @@ import dataclasses import datetime -import typing from ._exceptions import NextcloudExceptionNotModified from ._misc import check_capabilities, nc_iso_time_to_datetime @@ -154,8 +153,8 @@ def available(self) -> bool: def get_activities( self, - filter_id: typing.Union[ActivityFilter, str] = "", - since: typing.Union[int, bool] = 0, + filter_id: ActivityFilter | str = "", + since: int | bool = 0, limit: int = 50, object_type: str = "", object_id: int = 0, diff --git a/nc_py_api/apps.py b/nc_py_api/apps.py index f38cf940..7124f29f 100644 --- a/nc_py_api/apps.py +++ b/nc_py_api/apps.py @@ -2,7 +2,6 @@ import dataclasses import datetime -import typing from ._misc import require_capabilities from ._session import NcSessionBasic @@ -77,7 +76,7 @@ def enable(self, app_id: str) -> None: raise ValueError("`app_id` parameter can not be empty") self._session.ocs("POST", f"{self._ep_base}/{app_id}") - def get_list(self, enabled: typing.Optional[bool] = None) -> list[str]: + def get_list(self, enabled: bool | None = None) -> list[str]: """Get the list of installed applications. :param enabled: filter to list all/only enabled/only disabled applications. diff --git a/nc_py_api/ex_app/integration_fastapi.py b/nc_py_api/ex_app/integration_fastapi.py index 0a60199b..298f2a40 100644 --- a/nc_py_api/ex_app/integration_fastapi.py +++ b/nc_py_api/ex_app/integration_fastapi.py @@ -54,11 +54,11 @@ def talk_bot_app(request: Request) -> TalkBotMessage: def set_handlers( fast_api_app: FastAPI, - enabled_handler: typing.Callable[[bool, NextcloudApp], typing.Union[str, typing.Awaitable[str]]], - heartbeat_handler: typing.Optional[typing.Callable[[], typing.Union[str, typing.Awaitable[str]]]] = None, - init_handler: typing.Optional[typing.Callable[[NextcloudApp], None]] = None, - models_to_fetch: typing.Optional[list[str]] = None, - models_download_params: typing.Optional[dict] = None, + enabled_handler: typing.Callable[[bool, NextcloudApp], str | typing.Awaitable[str]], + heartbeat_handler: typing.Callable[[], str | typing.Awaitable[str]] | None = None, + init_handler: typing.Callable[[NextcloudApp], None] | None = None, + models_to_fetch: list[str] | None = None, + models_download_params: dict | None = None, map_app_static: bool = True, ): """Defines handlers for the application. @@ -133,7 +133,7 @@ def __map_app_static_folders(fast_api_app: FastAPI): def __fetch_models_task( nc: NextcloudApp, - init_handler: typing.Optional[typing.Callable[[NextcloudApp], None]], + init_handler: typing.Callable[[NextcloudApp], None] | None, models: list[str], params: dict[str, typing.Any], ) -> None: diff --git a/nc_py_api/ex_app/misc.py b/nc_py_api/ex_app/misc.py index f609ccd7..683e4762 100644 --- a/nc_py_api/ex_app/misc.py +++ b/nc_py_api/ex_app/misc.py @@ -1,7 +1,6 @@ """Different miscellaneous optimization/helper functions for the Nextcloud Applications.""" import os -import typing from sys import platform @@ -26,7 +25,7 @@ def _get_app_cache_dir() -> str: return r -def verify_version(finalize_update: bool = True) -> typing.Optional[tuple[str, str]]: +def verify_version(finalize_update: bool = True) -> tuple[str, str] | None: """Returns tuple with an old version and new version or ``None`` if there was no update taken. :param finalize_update: Flag indicating whether update information should be updated. diff --git a/nc_py_api/ex_app/ui/files_actions.py b/nc_py_api/ex_app/ui/files_actions.py index cf1475cd..fa9f4fb5 100644 --- a/nc_py_api/ex_app/ui/files_actions.py +++ b/nc_py_api/ex_app/ui/files_actions.py @@ -3,7 +3,6 @@ import dataclasses import datetime import os -import typing from pydantic import BaseModel @@ -87,11 +86,11 @@ class UiActionFileInfo(BaseModel): """Last modified time""" userId: str """The ID of the user performing the action.""" - shareOwner: typing.Optional[str] + shareOwner: str | None """If the object is shared, this is a display name of the share owner.""" - shareOwnerId: typing.Optional[str] + shareOwnerId: str | None """If the object is shared, this is the owner ID of the share.""" - instanceId: typing.Optional[str] + instanceId: str | None """Nextcloud instance ID.""" def to_fs_node(self) -> FsNode: @@ -150,7 +149,7 @@ def unregister(self, name: str, not_fail=True) -> None: if not not_fail: raise e from None - def get_entry(self, name: str) -> typing.Optional[UiFileActionEntry]: + def get_entry(self, name: str) -> UiFileActionEntry | None: """Get information of the file action meny entry for current app.""" require_capabilities("app_api", self._session.capabilities) try: diff --git a/nc_py_api/ex_app/ui/resources.py b/nc_py_api/ex_app/ui/resources.py index aae0df7e..82c3bc42 100644 --- a/nc_py_api/ex_app/ui/resources.py +++ b/nc_py_api/ex_app/ui/resources.py @@ -1,7 +1,6 @@ """API for adding scripts, styles, initial-states to the Nextcloud UI.""" import dataclasses -import typing from ..._exceptions import NextcloudExceptionNotFound from ..._misc import require_capabilities @@ -40,7 +39,7 @@ def key(self) -> str: return self._raw_data["key"] @property - def value(self) -> typing.Union[dict, list]: + def value(self) -> dict | list: """Object for the page(template).""" return self._raw_data["value"] @@ -87,7 +86,7 @@ class _UiResources: def __init__(self, session: NcSessionApp): self._session = session - def set_initial_state(self, ui_type: str, name: str, key: str, value: typing.Union[dict, list]) -> None: + def set_initial_state(self, ui_type: str, name: str, key: str, value: dict | list) -> None: """Add or update initial state for the page(template).""" require_capabilities("app_api", self._session.capabilities) params = { @@ -111,7 +110,7 @@ def delete_initial_state(self, ui_type: str, name: str, key: str, not_fail=True) if not not_fail: raise e from None - def get_initial_state(self, ui_type: str, name: str, key: str) -> typing.Optional[UiInitState]: + def get_initial_state(self, ui_type: str, name: str, key: str) -> UiInitState | None: """Get information about initial state for the page(template) by object name.""" require_capabilities("app_api", self._session.capabilities) try: @@ -149,7 +148,7 @@ def delete_script(self, ui_type: str, name: str, path: str, not_fail=True) -> No if not not_fail: raise e from None - def get_script(self, ui_type: str, name: str, path: str) -> typing.Optional[UiScript]: + def get_script(self, ui_type: str, name: str, path: str) -> UiScript | None: """Get information about script for the page(template) by object name.""" require_capabilities("app_api", self._session.capabilities) try: @@ -186,7 +185,7 @@ def delete_style(self, ui_type: str, name: str, path: str, not_fail=True) -> Non if not not_fail: raise e from None - def get_style(self, ui_type: str, name: str, path: str) -> typing.Optional[UiStyle]: + def get_style(self, ui_type: str, name: str, path: str) -> UiStyle | None: """Get information about style(css) for the page(template) by object name.""" require_capabilities("app_api", self._session.capabilities) try: diff --git a/nc_py_api/ex_app/ui/top_menu.py b/nc_py_api/ex_app/ui/top_menu.py index 1b15ccce..13d410eb 100644 --- a/nc_py_api/ex_app/ui/top_menu.py +++ b/nc_py_api/ex_app/ui/top_menu.py @@ -1,7 +1,6 @@ """Nextcloud API for working with Top App menu.""" import dataclasses -import typing from ..._exceptions import NextcloudExceptionNotFound from ..._misc import require_capabilities @@ -78,7 +77,7 @@ def unregister(self, name: str, not_fail=True) -> None: if not not_fail: raise e from None - def get_entry(self, name: str) -> typing.Optional[UiTopMenuEntry]: + def get_entry(self, name: str) -> UiTopMenuEntry | None: """Get information of the top meny entry for current app.""" require_capabilities("app_api", self._session.capabilities) try: diff --git a/nc_py_api/ex_app/uvicorn_fastapi.py b/nc_py_api/ex_app/uvicorn_fastapi.py index 40e64c2c..8443f4a5 100644 --- a/nc_py_api/ex_app/uvicorn_fastapi.py +++ b/nc_py_api/ex_app/uvicorn_fastapi.py @@ -12,7 +12,7 @@ def run_app( - uvicorn_app: typing.Union[typing.Callable, str, typing.Any], + uvicorn_app: typing.Callable | str | typing.Any, *args, **kwargs, ) -> None: diff --git a/nc_py_api/files/__init__.py b/nc_py_api/files/__init__.py index 90f1b869..4e04f7ec 100644 --- a/nc_py_api/files/__init__.py +++ b/nc_py_api/files/__init__.py @@ -4,7 +4,6 @@ import datetime import email.utils import enum -import typing from .. import _misc @@ -36,7 +35,7 @@ def __init__(self, **kwargs): self.last_modified = kwargs.get("last_modified", datetime.datetime(1970, 1, 1)) except (ValueError, TypeError): self.last_modified = datetime.datetime(1970, 1, 1) - self._trashbin: dict[str, typing.Union[str, int]] = {} + self._trashbin: dict[str, str | int] = {} for i in ("trashbin_filename", "trashbin_original_location", "trashbin_deletion_time"): if i in kwargs: self._trashbin[i] = kwargs[i] @@ -70,7 +69,7 @@ def last_modified(self) -> datetime.datetime: return self._last_modified @last_modified.setter - def last_modified(self, value: typing.Union[str, datetime.datetime]): + def last_modified(self, value: str | datetime.datetime): if isinstance(value, str): self._last_modified = email.utils.parsedate_to_datetime(value) else: diff --git a/nc_py_api/files/files.py b/nc_py_api/files/files.py index 25f207eb..1217b849 100644 --- a/nc_py_api/files/files.py +++ b/nc_py_api/files/files.py @@ -6,7 +6,6 @@ from io import BytesIO from json import dumps, loads from pathlib import Path -from typing import Optional, Union from urllib.parse import unquote from xml.etree import ElementTree @@ -73,7 +72,7 @@ def __init__(self, session: NcSessionBasic): self._session = session self.sharing = _FilesSharingAPI(session) - def listdir(self, path: Union[str, FsNode] = "", depth: int = 1, exclude_self=True) -> list[FsNode]: + def listdir(self, path: str | FsNode = "", depth: int = 1, exclude_self=True) -> list[FsNode]: """Returns a list of all entries in the specified directory. :param path: path to the directory to get the list. @@ -87,7 +86,7 @@ def listdir(self, path: Union[str, FsNode] = "", depth: int = 1, exclude_self=Tr path = path.user_path if isinstance(path, FsNode) else path return self._listdir(self._session.user, path, properties=properties, depth=depth, exclude_self=exclude_self) - def by_id(self, file_id: Union[int, str, FsNode]) -> Optional[FsNode]: + def by_id(self, file_id: int | str | FsNode) -> FsNode | None: """Returns :py:class:`~nc_py_api.files.FsNode` by file_id if any. :param file_id: can be full file ID with Nextcloud instance ID or only clear file ID. @@ -96,13 +95,13 @@ def by_id(self, file_id: Union[int, str, FsNode]) -> Optional[FsNode]: result = self.find(req=["eq", "fileid", file_id]) return result[0] if result else None - def by_path(self, path: Union[str, FsNode]) -> Optional[FsNode]: + def by_path(self, path: str | FsNode) -> FsNode | None: """Returns :py:class:`~nc_py_api.files.FsNode` by exact path if any.""" path = path.user_path if isinstance(path, FsNode) else path result = self.listdir(path, depth=0, exclude_self=False) return result[0] if result else None - def find(self, req: list, path: Union[str, FsNode] = "") -> list[FsNode]: + def find(self, req: list, path: str | FsNode = "") -> list[FsNode]: """Searches a directory for a file or subdirectory with a name. :param req: list of conditions to search for. Detailed description here... @@ -132,7 +131,7 @@ def find(self, req: list, path: Union[str, FsNode] = "") -> list[FsNode]: request_info = f"find: {self._session.user}, {req}, {path}" return self._lf_parse_webdav_response(webdav_response, request_info) - def download(self, path: Union[str, FsNode]) -> bytes: + def download(self, path: str | FsNode) -> bytes: """Downloads and returns the content of a file. :param path: path to download file. @@ -142,7 +141,7 @@ def download(self, path: Union[str, FsNode]) -> bytes: check_error(response, f"download: user={self._session.user}, path={path}") return response.content - def download2stream(self, path: Union[str, FsNode], fp, **kwargs) -> None: + def download2stream(self, path: str | FsNode, fp, **kwargs) -> None: """Downloads file to the given `fp` object. :param path: path to download file. @@ -151,7 +150,7 @@ def download2stream(self, path: Union[str, FsNode], fp, **kwargs) -> None: :param kwargs: **chunk_size** an int value specifying chunk size to write. Default = **5Mb** """ path = path.user_path if isinstance(path, FsNode) else path - if isinstance(fp, (str, Path)): + if isinstance(fp, str | Path): with builtins.open(fp, "wb") as f: self.__download2stream(path, f, **kwargs) elif hasattr(fp, "write"): @@ -159,9 +158,7 @@ def download2stream(self, path: Union[str, FsNode], fp, **kwargs) -> None: else: raise TypeError("`fp` must be a path to file or an object with `write` method.") - def download_directory_as_zip( - self, path: Union[str, FsNode], local_path: Union[str, Path, None] = None, **kwargs - ) -> Path: + def download_directory_as_zip(self, path: str | FsNode, local_path: str | Path | None = None, **kwargs) -> Path: """Downloads a remote directory as zip archive. :param path: path to directory to download. @@ -184,7 +181,7 @@ def download_directory_as_zip( fp.write(data_chunk) return Path(result_path) - def upload(self, path: Union[str, FsNode], content: Union[bytes, str]) -> FsNode: + def upload(self, path: str | FsNode, content: bytes | str) -> FsNode: """Creates a file with the specified content at the specified path. :param path: file's upload path. @@ -196,7 +193,7 @@ def upload(self, path: Union[str, FsNode], content: Union[bytes, str]) -> FsNode check_error(response, f"upload: user={self._session.user}, path={path}, size={len(content)}") return FsNode(full_path.strip("/"), **self.__get_etag_fileid_from_response(response)) - def upload_stream(self, path: Union[str, FsNode], fp, **kwargs) -> FsNode: + def upload_stream(self, path: str | FsNode, fp, **kwargs) -> FsNode: """Creates a file with content provided by `fp` object at the specified path. :param path: file's upload path. @@ -206,7 +203,7 @@ def upload_stream(self, path: Union[str, FsNode], fp, **kwargs) -> FsNode: """ path = path.user_path if isinstance(path, FsNode) else path chunk_size = kwargs.get("chunk_size", 5 * 1024 * 1024) - if isinstance(fp, (str, Path)): + if isinstance(fp, str | Path): with builtins.open(fp, "rb") as f: return self.__upload_stream(path, f, chunk_size) elif hasattr(fp, "read"): @@ -214,7 +211,7 @@ def upload_stream(self, path: Union[str, FsNode], fp, **kwargs) -> FsNode: else: raise TypeError("`fp` must be a path to file or an object with `read` method.") - def mkdir(self, path: Union[str, FsNode]) -> FsNode: + def mkdir(self, path: str | FsNode) -> FsNode: """Creates a new directory. :param path: path of the directory to be created. @@ -226,7 +223,7 @@ def mkdir(self, path: Union[str, FsNode]) -> FsNode: full_path += "/" if not full_path.endswith("/") else "" return FsNode(full_path.lstrip("/"), **self.__get_etag_fileid_from_response(response)) - def makedirs(self, path: Union[str, FsNode], exist_ok=False) -> Optional[FsNode]: + def makedirs(self, path: str | FsNode, exist_ok=False) -> FsNode | None: """Creates a new directory and subdirectories. :param path: path of the directories to be created. @@ -249,7 +246,7 @@ def makedirs(self, path: Union[str, FsNode], exist_ok=False) -> Optional[FsNode] raise e from None return result - def delete(self, path: Union[str, FsNode], not_fail=False) -> None: + def delete(self, path: str | FsNode, not_fail=False) -> None: """Deletes a file/directory (moves to trash if trash is enabled). :param path: path to delete. @@ -261,7 +258,7 @@ def delete(self, path: Union[str, FsNode], not_fail=False) -> None: return check_error(response) - def move(self, path_src: Union[str, FsNode], path_dest: Union[str, FsNode], overwrite=False) -> FsNode: + def move(self, path_src: str | FsNode, path_dest: str | FsNode, overwrite=False) -> FsNode: """Moves an existing file or a directory. :param path_src: path of an existing file/directory. @@ -283,7 +280,7 @@ def move(self, path_src: Union[str, FsNode], path_dest: Union[str, FsNode], over check_error(response, f"move: user={self._session.user}, src={path_src}, dest={dest}, {overwrite}") return self.find(req=["eq", "fileid", response.headers["OC-FileId"]])[0] - def copy(self, path_src: Union[str, FsNode], path_dest: Union[str, FsNode], overwrite=False) -> FsNode: + def copy(self, path_src: str | FsNode, path_dest: str | FsNode, overwrite=False) -> FsNode: """Copies an existing file/directory. :param path_src: path of an existing file/directory. @@ -306,7 +303,7 @@ def copy(self, path_src: Union[str, FsNode], path_dest: Union[str, FsNode], over return self.find(req=["eq", "fileid", response.headers["OC-FileId"]])[0] def list_by_criteria( - self, properties: Optional[list[str]] = None, tags: Optional[list[Union[int, SystemTag]]] = None + self, properties: list[str] | None = None, tags: list[int | SystemTag] | None = None ) -> list[FsNode]: """Returns a list of all files/directories for the current user filtered by the specified values. @@ -337,7 +334,7 @@ def list_by_criteria( check_error(webdav_response, request_info) return self._lf_parse_webdav_response(webdav_response, request_info) - def setfav(self, path: Union[str, FsNode], value: Union[int, bool]) -> None: + def setfav(self, path: str | FsNode, value: int | bool) -> None: """Sets or unsets favourite flag for specific file. :param path: path to the object to set the state. @@ -364,7 +361,7 @@ def trashbin_list(self) -> list[FsNode]: self._session.user, "", properties=properties, depth=1, exclude_self=False, prop_type=PropFindType.TRASHBIN ) - def trashbin_restore(self, path: Union[str, FsNode]) -> None: + def trashbin_restore(self, path: str | FsNode) -> None: """Restore a file/directory from the TrashBin. :param path: path to delete, e.g., the ``user_path`` field from ``FsNode`` or the **FsNode** class itself. @@ -381,7 +378,7 @@ def trashbin_restore(self, path: Union[str, FsNode]) -> None: ) check_error(response, f"trashbin_restore: user={self._session.user}, src={path}, dest={dest}") - def trashbin_delete(self, path: Union[str, FsNode], not_fail=False) -> None: + def trashbin_delete(self, path: str | FsNode, not_fail=False) -> None: """Deletes a file/directory permanently from the TrashBin. :param path: path to delete, e.g., the ``user_path`` field from ``FsNode`` or the **FsNode** class itself. @@ -463,10 +460,10 @@ def create_tag(self, name: str, user_visible: bool = True, user_assignable: bool def update_tag( self, - tag_id: Union[int, SystemTag], - name: Optional[str] = None, - user_visible: Optional[bool] = None, - user_assignable: Optional[bool] = None, + tag_id: int | SystemTag, + name: str | None = None, + user_visible: bool | None = None, + user_assignable: bool | None = None, ) -> None: """Updates the Tag information.""" tag_id = tag_id.tag_id if isinstance(tag_id, SystemTag) else tag_id @@ -494,7 +491,7 @@ def update_tag( ) check_error(response) - def delete_tag(self, tag_id: Union[int, SystemTag]) -> None: + def delete_tag(self, tag_id: int | SystemTag) -> None: """Deletes the tag.""" tag_id = tag_id.tag_id if isinstance(tag_id, SystemTag) else tag_id response = self._session.adapter_dav.delete(f"/systemtags/{tag_id}") @@ -507,17 +504,15 @@ def tag_by_name(self, tag_name: str) -> SystemTag: raise NextcloudExceptionNotFound(f"Tag with name='{tag_name}' not found.") return r[0] - def assign_tag(self, file_id: Union[FsNode, int], tag_id: Union[SystemTag, int]) -> None: + def assign_tag(self, file_id: FsNode | int, tag_id: SystemTag | int) -> None: """Assigns Tag to a file/directory.""" self._file_change_tag_state(file_id, tag_id, True) - def unassign_tag(self, file_id: Union[FsNode, int], tag_id: Union[SystemTag, int]) -> None: + def unassign_tag(self, file_id: FsNode | int, tag_id: SystemTag | int) -> None: """Removes Tag from a file/directory.""" self._file_change_tag_state(file_id, tag_id, False) - def _file_change_tag_state( - self, file_id: Union[FsNode, int], tag_id: Union[SystemTag, int], tag_state: bool - ) -> None: + def _file_change_tag_state(self, file_id: FsNode | int, tag_id: SystemTag | int, tag_state: bool) -> None: fs_object = file_id.info.fileid if isinstance(file_id, FsNode) else file_id tag = tag_id.tag_id if isinstance(tag_id, SystemTag) else tag_id response = self._session.adapter_dav.request( diff --git a/nc_py_api/files/sharing.py b/nc_py_api/files/sharing.py index 6be0fefb..ea2ce38b 100644 --- a/nc_py_api/files/sharing.py +++ b/nc_py_api/files/sharing.py @@ -1,7 +1,5 @@ """Nextcloud API for working with the files shares.""" -import typing - from .. import _misc, _session from . import FilePermissions, FsNode, Share, ShareType @@ -19,9 +17,7 @@ def available(self) -> bool: """Returns True if the Nextcloud instance supports this feature, False otherwise.""" return not _misc.check_capabilities("files_sharing.api_enabled", self._session.capabilities) - def get_list( - self, shared_with_me=False, reshares=False, subfiles=False, path: typing.Union[str, FsNode] = "" - ) -> list[Share]: + def get_list(self, shared_with_me=False, reshares=False, subfiles=False, path: str | FsNode = "") -> list[Share]: """Returns lists of shares. :param shared_with_me: Shares should be with the current user. @@ -55,9 +51,9 @@ def get_inherited(self, path: str) -> list[Share]: def create( self, - path: typing.Union[str, FsNode], + path: str | FsNode, share_type: ShareType, - permissions: typing.Optional[FilePermissions] = None, + permissions: FilePermissions | None = None, share_with: str = "", **kwargs, ) -> Share: @@ -104,7 +100,7 @@ def create( params["label"] = kwargs["label"] return Share(self._session.ocs("POST", f"{self._ep_base}/shares", params=params)) - def update(self, share_id: typing.Union[int, Share], **kwargs) -> Share: + def update(self, share_id: int | Share, **kwargs) -> Share: """Updates the share options. :param share_id: ID of the Share to update. @@ -130,7 +126,7 @@ def update(self, share_id: typing.Union[int, Share], **kwargs) -> Share: params["label"] = kwargs["label"] return Share(self._session.ocs("PUT", f"{self._ep_base}/shares/{share_id}", params=params)) - def delete(self, share_id: typing.Union[int, Share]) -> None: + def delete(self, share_id: int | Share) -> None: """Removes the given share. :param share_id: The Share object or an ID of the share. @@ -143,13 +139,13 @@ def get_pending(self) -> list[Share]: """Returns all pending shares for current user.""" return [Share(i) for i in self._session.ocs("GET", f"{self._ep_base}/shares/pending")] - def accept_share(self, share_id: typing.Union[int, Share]) -> None: + def accept_share(self, share_id: int | Share) -> None: """Accept pending share.""" _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities) share_id = share_id.share_id if isinstance(share_id, Share) else share_id self._session.ocs("POST", f"{self._ep_base}/pending/{share_id}") - def decline_share(self, share_id: typing.Union[int, Share]) -> None: + def decline_share(self, share_id: int | Share) -> None: """Decline pending share.""" _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities) share_id = share_id.share_id if isinstance(share_id, Share) else share_id @@ -160,7 +156,7 @@ def get_deleted(self) -> list[Share]: _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities) return [Share(i) for i in self._session.ocs("GET", f"{self._ep_base}/deletedshares")] - def undelete(self, share_id: typing.Union[int, Share]) -> None: + def undelete(self, share_id: int | Share) -> None: """Undelete a deleted share.""" _misc.require_capabilities("files_sharing.api_enabled", self._session.capabilities) share_id = share_id.share_id if isinstance(share_id, Share) else share_id diff --git a/nc_py_api/nextcloud.py b/nc_py_api/nextcloud.py index 65477d60..512db972 100644 --- a/nc_py_api/nextcloud.py +++ b/nc_py_api/nextcloud.py @@ -1,7 +1,6 @@ """Nextcloud class providing access to all API endpoints.""" from abc import ABC -from typing import Optional, Union from fastapi import Request from httpx import Headers as HttpxHeaders @@ -78,7 +77,7 @@ def srv_version(self) -> ServerVersion: """Returns dictionary with the server version.""" return self._session.nc_version - def check_capabilities(self, capabilities: Union[str, list[str]]) -> list[str]: + def check_capabilities(self, capabilities: str | list[str]) -> list[str]: """Returns the list with missing capabilities if any. :param capabilities: one or more features to check for. @@ -98,7 +97,7 @@ def response_headers(self) -> HttpxHeaders: return self._session.response_headers @property - def theme(self) -> Optional[ThemingInfo]: + def theme(self) -> ThemingInfo | None: """Returns Theme information.""" return get_parsed_theme(self.capabilities["theming"]) if "theming" in self.capabilities else None diff --git a/nc_py_api/notes.py b/nc_py_api/notes.py index ba5b53ee..71ba57e7 100644 --- a/nc_py_api/notes.py +++ b/nc_py_api/notes.py @@ -109,10 +109,10 @@ def available(self) -> bool: def get_list( self, - category: typing.Optional[str] = None, - modified_since: typing.Optional[int] = None, - limit: typing.Optional[int] = None, - cursor: typing.Optional[str] = None, + category: str | None = None, + modified_since: int | None = None, + limit: int | None = None, + cursor: str | None = None, no_content: bool = False, etag: bool = False, ) -> list[Note]: @@ -157,10 +157,10 @@ def by_id(self, note: Note) -> Note: def create( self, title: str, - content: typing.Optional[str] = None, - category: typing.Optional[str] = None, - favorite: typing.Optional[bool] = None, - last_modified: typing.Optional[typing.Union[int, str, datetime.datetime]] = None, + content: str | None = None, + category: str | None = None, + favorite: bool | None = None, + last_modified: int | str | datetime.datetime | None = None, ) -> Note: """Create new Note.""" require_capabilities("notes", self._session.capabilities) @@ -177,10 +177,10 @@ def create( def update( self, note: Note, - title: typing.Optional[str] = None, - content: typing.Optional[str] = None, - category: typing.Optional[str] = None, - favorite: typing.Optional[bool] = None, + title: str | None = None, + content: str | None = None, + category: str | None = None, + favorite: bool | None = None, overwrite: bool = False, ) -> Note: """Updates Note. @@ -204,7 +204,7 @@ def update( ) ) - def delete(self, note: typing.Union[int, Note]) -> None: + def delete(self, note: int | Note) -> None: """Deletes a Note. :param note: note id or :py:class:`~nc_py_api.notes.Note`. @@ -219,7 +219,7 @@ def get_settings(self) -> NotesSettings: r = self.__response_to_json(self._session.adapter.get(self._ep_base + "/settings")) return {"notes_path": r["notesPath"], "file_suffix": r["fileSuffix"]} - def set_settings(self, notes_path: typing.Optional[str] = None, file_suffix: typing.Optional[str] = None) -> None: + def set_settings(self, notes_path: str | None = None, file_suffix: str | None = None) -> None: """Change specified setting(s).""" if notes_path is None and file_suffix is None: raise ValueError("No setting to change.") diff --git a/nc_py_api/notifications.py b/nc_py_api/notifications.py index acbf9577..ac81f6d4 100644 --- a/nc_py_api/notifications.py +++ b/nc_py_api/notifications.py @@ -2,7 +2,6 @@ import dataclasses import datetime -import typing from ._misc import ( check_capabilities, @@ -94,8 +93,8 @@ def create( self, subject: str, message: str = "", - subject_params: typing.Optional[dict] = None, - message_params: typing.Optional[dict] = None, + subject_params: dict | None = None, + message_params: dict | None = None, link: str = "", ) -> str: """Create a Notification for the current user and returns it's ObjectID. @@ -138,7 +137,7 @@ def get_one(self, notification_id: int) -> Notification: require_capabilities("notifications", self._session.capabilities) return Notification(self._session.ocs("GET", f"{self._ep_base}/{notification_id}")) - def by_object_id(self, object_id: str) -> typing.Optional[Notification]: + def by_object_id(self, object_id: str) -> Notification | None: """Returns Notification if any by its object ID. .. note:: this method is a temporary workaround until `create` can return `notification_id`. diff --git a/nc_py_api/options.py b/nc_py_api/options.py index 7f19f7e2..043be956 100644 --- a/nc_py_api/options.py +++ b/nc_py_api/options.py @@ -4,7 +4,6 @@ Specifying options in **kwargs** has higher priority than this. """ -import typing from os import environ from dotenv import load_dotenv @@ -14,21 +13,21 @@ XDEBUG_SESSION = environ.get("XDEBUG_SESSION", "") """Dev option, for debugging PHP code.""" -NPA_TIMEOUT: typing.Optional[int] +NPA_TIMEOUT: int | None """Default timeout for OCS API calls. Set to ``None`` to disable timeouts for development.""" try: NPA_TIMEOUT = int(environ.get("NPA_TIMEOUT", 30)) except (TypeError, ValueError): NPA_TIMEOUT = None -NPA_TIMEOUT_DAV: typing.Optional[int] +NPA_TIMEOUT_DAV: int | None """File operations timeout, usually it is OCS timeout multiplied by 3.""" try: NPA_TIMEOUT_DAV = int(environ.get("NPA_TIMEOUT_DAV", 30 * 3)) except (TypeError, ValueError): NPA_TIMEOUT_DAV = None -NPA_NC_CERT: typing.Union[bool, str] +NPA_NC_CERT: bool | str """Option to enable/disable Nextcloud certificate verification. SSL certificates (a.k.a CA bundle) used to verify the identity of requested hosts. Either **True** (default CA bundle), diff --git a/nc_py_api/talk.py b/nc_py_api/talk.py index 71fb50a8..73f997b1 100644 --- a/nc_py_api/talk.py +++ b/nc_py_api/talk.py @@ -4,7 +4,6 @@ import datetime import enum import os -import typing from . import files as _files @@ -568,7 +567,7 @@ def last_common_read_message(self) -> int: return self._raw_data["lastCommonReadMessage"] @property - def last_message(self) -> typing.Optional[TalkMessage]: + def last_message(self) -> TalkMessage | None: """Last message in a conversation if available, otherwise ``empty``. .. note:: Even when given, the message will not contain the ``parent`` or ``reactionsSelf`` @@ -625,7 +624,7 @@ def recording_status(self) -> CallRecordingStatus: return CallRecordingStatus(self._raw_data.get("callRecording", CallRecordingStatus.NO_RECORDING)) @property - def status_clear_at(self) -> typing.Optional[int]: + def status_clear_at(self) -> int | None: """Unix Timestamp representing the time to clear the status. .. note:: Available only for ``one-to-one`` conversations. @@ -760,7 +759,7 @@ def last_error_date(self) -> int: return self._raw_data["last_error_date"] @property - def last_error_message(self) -> typing.Optional[str]: + def last_error_message(self) -> str | None: """The last exception message or error response information when trying to reach the bot.""" return self._raw_data["last_error_message"] diff --git a/nc_py_api/talk_bot.py b/nc_py_api/talk_bot.py index 16b99a1a..943a1017 100644 --- a/nc_py_api/talk_bot.py +++ b/nc_py_api/talk_bot.py @@ -111,7 +111,7 @@ def enabled_handler(self, enabled: bool, nc: NextcloudApp) -> None: nc.unregister_talk_bot(self.callback_url) def send_message( - self, message: str, reply_to_message: typing.Union[int, TalkBotMessage], silent: bool = False, token: str = "" + self, message: str, reply_to_message: int | TalkBotMessage, silent: bool = False, token: str = "" ) -> tuple[httpx.Response, str]: """Send a message and returns a "reference string" to identify the message again in a "get messages" request. @@ -138,9 +138,7 @@ def send_message( } return self._sign_send_request("POST", f"/{token}/message", params, message), reference_id - def react_to_message( - self, message: typing.Union[int, TalkBotMessage], reaction: str, token: str = "" - ) -> httpx.Response: + def react_to_message(self, message: int | TalkBotMessage, reaction: str, token: str = "") -> httpx.Response: """React to a message. :param message: Message ID or :py:class:`~nc_py_api.talk_bot.TalkBotMessage` to react to. @@ -159,9 +157,7 @@ def react_to_message( } return self._sign_send_request("POST", f"/{token}/reaction/{message_id}", params, reaction) - def delete_reaction( - self, message: typing.Union[int, TalkBotMessage], reaction: str, token: str = "" - ) -> httpx.Response: + def delete_reaction(self, message: int | TalkBotMessage, reaction: str, token: str = "") -> httpx.Response: """Removes reaction from a message. :param message: Message ID or :py:class:`~nc_py_api.talk_bot.TalkBotMessage` to remove reaction from. @@ -204,7 +200,7 @@ def _sign_send_request(self, method: str, url_suffix: str, data: dict, data_to_s ) -def get_bot_secret(callback_url: str) -> typing.Union[bytes, None]: +def get_bot_secret(callback_url: str) -> bytes | None: """Returns the bot's secret from an environment variable or from the application's configuration on the server.""" sha_1 = hashlib.sha1(usedforsecurity=False) string_to_hash = os.environ["APP_ID"] + "_" + callback_url diff --git a/nc_py_api/user_status.py b/nc_py_api/user_status.py index bd10976d..4f984799 100644 --- a/nc_py_api/user_status.py +++ b/nc_py_api/user_status.py @@ -14,7 +14,7 @@ class ClearAt: clear_type: str """Possible values: ``period``, ``end-of``""" - time: typing.Union[str, int] + time: str | int """Depending of ``type`` it can be number of seconds relative to ``now`` or one of the next values: ``day``""" def __init__(self, raw_data: dict): @@ -32,7 +32,7 @@ class PredefinedStatus: """Icon in string(UTF) format""" message: str """The message defined for this status. It is translated, so it depends on the user's language setting.""" - clear_at: typing.Optional[ClearAt] + clear_at: ClearAt | None """When the default, if not override, the predefined status will be cleared.""" def __init__(self, raw_status: dict): @@ -68,7 +68,7 @@ def status_icon(self) -> str: return self._raw_data.get("icon", "") @property - def status_clear_at(self) -> typing.Optional[int]: + def status_clear_at(self) -> int | None: """Unix Timestamp representing the time to clear the status.""" return self._raw_data.get("clearAt", None) @@ -86,7 +86,7 @@ class CurrentUserStatus(UserStatus): """Information about current user status.""" @property - def status_id(self) -> typing.Optional[str]: + def status_id(self) -> str | None: """ID of the predefined status.""" return self._raw_data["messageId"] @@ -120,7 +120,7 @@ def available(self) -> bool: """Returns True if the Nextcloud instance supports this feature, False otherwise.""" return not check_capabilities("user_status.enabled", self._session.capabilities) - def get_list(self, limit: typing.Optional[int] = None, offset: typing.Optional[int] = None) -> list[UserStatus]: + def get_list(self, limit: int | None = None, offset: int | None = None) -> list[UserStatus]: """Returns statuses for all users.""" require_capabilities("user_status.enabled", self._session.capabilities) data = kwargs_to_params(["limit", "offset"], limit=limit, offset=offset) @@ -132,7 +132,7 @@ def get_current(self) -> CurrentUserStatus: require_capabilities("user_status.enabled", self._session.capabilities) return CurrentUserStatus(self._session.ocs("GET", f"{self._ep_base}/user_status")) - def get(self, user_id: str) -> typing.Optional[UserStatus]: + def get(self, user_id: str) -> UserStatus | None: """Returns the user status for the specified user.""" require_capabilities("user_status.enabled", self._session.capabilities) try: @@ -157,7 +157,7 @@ def set_predefined(self, status_id: str, clear_at: int = 0) -> None: if self._session.nc_version["major"] < 27: return require_capabilities("user_status.enabled", self._session.capabilities) - params: dict[str, typing.Union[int, str]] = {"messageId": status_id} + params: dict[str, int | str] = {"messageId": status_id} if clear_at: params["clearAt"] = clear_at self._session.ocs("PUT", f"{self._ep_base}/user_status/message/predefined", params=params) @@ -167,7 +167,7 @@ def set_status_type(self, value: typing.Literal["online", "away", "dnd", "invisi require_capabilities("user_status.enabled", self._session.capabilities) self._session.ocs("PUT", f"{self._ep_base}/user_status/status", params={"statusType": value}) - def set_status(self, message: typing.Optional[str] = None, clear_at: int = 0, status_icon: str = "") -> None: + def set_status(self, message: str | None = None, clear_at: int = 0, status_icon: str = "") -> None: """Sets current user status. :param message: Message text to set in the status. @@ -180,14 +180,14 @@ def set_status(self, message: typing.Optional[str] = None, clear_at: int = 0, st return if status_icon: require_capabilities("user_status.supports_emoji", self._session.capabilities) - params: dict[str, typing.Union[int, str]] = {"message": message} + params: dict[str, int | str] = {"message": message} if clear_at: params["clearAt"] = clear_at if status_icon: params["statusIcon"] = status_icon self._session.ocs("PUT", f"{self._ep_base}/user_status/message/custom", params=params) - def get_backup_status(self, user_id: str = "") -> typing.Optional[UserStatus]: + def get_backup_status(self, user_id: str = "") -> UserStatus | None: """Get the backup status of the user if any.""" require_capabilities("user_status.enabled", self._session.capabilities) user_id = user_id if user_id else self._session.user @@ -195,7 +195,7 @@ def get_backup_status(self, user_id: str = "") -> typing.Optional[UserStatus]: raise ValueError("user_id can not be empty.") return self.get(f"_{user_id}") - def restore_backup_status(self, status_id: str) -> typing.Optional[CurrentUserStatus]: + def restore_backup_status(self, status_id: str) -> CurrentUserStatus | None: """Restores the backup state as current for the current user.""" require_capabilities("user_status.enabled", self._session.capabilities) require_capabilities("user_status.restore", self._session.capabilities) diff --git a/nc_py_api/users.py b/nc_py_api/users.py index 62da37cb..dcca1dc5 100644 --- a/nc_py_api/users.py +++ b/nc_py_api/users.py @@ -167,9 +167,7 @@ class _UsersAPI: def __init__(self, session: NcSessionBasic): self._session = session - def get_list( - self, mask: typing.Optional[str] = "", limit: typing.Optional[int] = None, offset: typing.Optional[int] = None - ) -> list[str]: + def get_list(self, mask: str | None = "", limit: int | None = None, offset: int | None = None) -> list[str]: """Returns list of user IDs.""" data = kwargs_to_params(["search", "limit", "offset"], search=mask, limit=limit, offset=offset) response_data = self._session.ocs("GET", self._ep_base, params=data) @@ -179,7 +177,7 @@ def get_user(self, user_id: str = "") -> UserInfo: """Returns detailed user information.""" return UserInfo(self._session.ocs("GET", f"{self._ep_base}/{user_id}" if user_id else "/ocs/v1.php/cloud/user")) - def create(self, user_id: str, display_name: typing.Optional[str] = None, **kwargs) -> None: + def create(self, user_id: str, display_name: str | None = None, **kwargs) -> None: """Create a new user on the Nextcloud server. :param user_id: id of the user to create. diff --git a/nc_py_api/users_groups.py b/nc_py_api/users_groups.py index f8a8e5d2..4db8c400 100644 --- a/nc_py_api/users_groups.py +++ b/nc_py_api/users_groups.py @@ -1,7 +1,6 @@ """Nextcloud API for working with user groups.""" import dataclasses -import typing from ._misc import kwargs_to_params from ._session import NcSessionBasic @@ -59,23 +58,21 @@ class _UsersGroupsAPI: def __init__(self, session: NcSessionBasic): self._session = session - def get_list( - self, mask: typing.Optional[str] = None, limit: typing.Optional[int] = None, offset: typing.Optional[int] = None - ) -> list[str]: + def get_list(self, mask: str | None = None, limit: int | None = None, offset: int | None = None) -> list[str]: """Returns a list of user groups IDs.""" data = kwargs_to_params(["search", "limit", "offset"], search=mask, limit=limit, offset=offset) response_data = self._session.ocs("GET", self._ep_base, params=data) return response_data["groups"] if response_data else [] def get_details( - self, mask: typing.Optional[str] = None, limit: typing.Optional[int] = None, offset: typing.Optional[int] = None + self, mask: str | None = None, limit: int | None = None, offset: int | None = None ) -> list[GroupDetails]: """Returns a list of user groups with detailed information.""" data = kwargs_to_params(["search", "limit", "offset"], search=mask, limit=limit, offset=offset) response_data = self._session.ocs("GET", f"{self._ep_base}/details", params=data) return [GroupDetails(i) for i in response_data["groups"]] if response_data else [] - def create(self, group_id: str, display_name: typing.Optional[str] = None) -> None: + def create(self, group_id: str, display_name: str | None = None) -> None: """Creates the users group.""" params = {"groupid": group_id} if display_name is not None: diff --git a/nc_py_api/weather_status.py b/nc_py_api/weather_status.py index 184e96ad..987e3f20 100644 --- a/nc_py_api/weather_status.py +++ b/nc_py_api/weather_status.py @@ -2,7 +2,6 @@ import dataclasses import enum -import typing from ._misc import check_capabilities, require_capabilities from ._session import NcSessionBasic @@ -61,9 +60,9 @@ def get_location(self) -> WeatherLocation: def set_location( self, - latitude: typing.Optional[float] = None, - longitude: typing.Optional[float] = None, - address: typing.Optional[str] = None, + latitude: float | None = None, + longitude: float | None = None, + address: str | None = None, ) -> bool: """Sets the user's location on the Nextcloud server. @@ -72,7 +71,7 @@ def set_location( :param address: city, index(*optional*) and country, e.g. "Paris, 75007, France" """ require_capabilities("weather_status.enabled", self._session.capabilities) - params: dict[str, typing.Union[str, float]] = {} + params: dict[str, str | float] = {} if latitude is not None and longitude is not None: params.update({"lat": latitude, "lon": longitude}) elif address: diff --git a/pyproject.toml b/pyproject.toml index 4731602a..8c9c5c59 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,7 +19,7 @@ license = "BSD-3-Clause" authors = [ { name = "Alexander Piskun", email = "bigcat88@icloud.com" }, ] -requires-python = ">=3.9" +requires-python = ">=3.10" classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", @@ -28,7 +28,6 @@ classifiers = [ "Operating System :: Microsoft :: Windows", "Operating System :: POSIX :: Linux", "Programming Language :: Python :: 3 :: Only", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", @@ -105,12 +104,12 @@ exclude = [ [tool.black] line-length = 120 -target-versions = ["py39"] +target-versions = ["py310"] preview = true [tool.ruff] line-length = 120 -target-version = "py39" +target-version = "py310" select = ["A", "B", "C", "D", "E", "F", "G", "I", "S", "SIM", "PIE", "Q", "RET", "RUF", "UP" , "W"] extend-ignore = ["D107", "D105", "D203", "D213", "D401", "I001", "RUF100"] @@ -131,7 +130,7 @@ max-complexity = 16 profile = "black" [tool.pylint] -master.py-version = "3.9" +master.py-version = "3.10" master.extension-pkg-allow-list = ["pydantic"] design.max-attributes = 8 design.max-locals = 16