options rework (#64)

Allows to specify `options` throw environment variables or `.env` file.

Also `options` now are independent in each `Nextcloud` or `NextcloudApp`
class, and this allows to specify them with `kwargs`

---------

Signed-off-by: Alexander Piskun <bigcat88@icloud.com>

Author: bigcat88

CHANGELOG.md

@@ -2,11 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
-## [0.0.27 - 2023-08-03]
+## [0.0.27 - 2023-08-05]
 
 ### Added
 
 - `Notifications API`
+- `options` now independent in each `Nextcloud` class. They can be specified in kwargs, environment or `.env` files.
 
 ### Changed
 

docs/Options.rst

@@ -6,6 +6,6 @@ Options
 -------
 
 .. autodata:: nc_py_api.options.XDEBUG_SESSION
-.. autodata:: nc_py_api.options.TIMEOUT
-.. autodata:: nc_py_api.options.TIMEOUT_DAV
-.. autodata:: nc_py_api.options.VERIFY_NC_CERTIFICATE
+.. autodata:: nc_py_api.options.NPA_TIMEOUT
+.. autodata:: nc_py_api.options.NPA_TIMEOUT_DAV
+.. autodata:: nc_py_api.options.NPA_NC_CERT

nc_py_api/_session.py

@@ -43,29 +43,51 @@ class ServerVersion(TypedDict):
     """Indicates if the subscription has extended support"""
 
 
+@dataclass
+class RuntimeOptions:
+    xdebug_session: str
+    timeout: Optional[int]
+    timeout_dav: Optional[int]
+    _nc_cert: Union[str, bool]
+
+    def __init__(self, **kwargs):
+        self.xdebug_session = kwargs.get("xdebug_session", options.XDEBUG_SESSION)
+        self.timeout = kwargs.get("npa_timeout", options.NPA_TIMEOUT)
+        self.timeout_dav = kwargs.get("npa_timeout_dav", options.NPA_TIMEOUT_DAV)
+        self._nc_cert = kwargs.get("npa_nc_cert", options.NPA_NC_CERT)
+
+    @property
+    def nc_cert(self) -> Union[str, bool]:
+        return self._nc_cert
+
+
 @dataclass
 class BasicConfig:
     endpoint: str
     dav_endpoint: str
     dav_url_suffix: str
+    options: RuntimeOptions
 
     def __init__(self, **kwargs):
-        full_nc_url = self._get_value("nextcloud_url", **kwargs)
+        full_nc_url = self._get_config_value("nextcloud_url", **kwargs)
         self.endpoint = full_nc_url.removesuffix("/index.php").removesuffix("/")
-        self.dav_url_suffix = self._get_value("dav_url_suffix", raise_not_found=False, **kwargs)
+        self.dav_url_suffix = self._get_config_value("dav_url_suffix", raise_not_found=False, **kwargs)
         if not self.dav_url_suffix:
             self.dav_url_suffix = "remote.php/dav"
         self.dav_url_suffix = "/" + self.dav_url_suffix.strip("/")
         self.dav_endpoint = self.endpoint + self.dav_url_suffix
+        self.options = RuntimeOptions(**kwargs)
 
     @staticmethod
-    def _get_value(value_name: str, raise_not_found=True, **kwargs):
-        value = kwargs.get(value_name, None)
-        if not value:
-            value = environ.get(value_name.upper(), None)
-        if not value and raise_not_found:
+    def _get_config_value(value_name: str, raise_not_found=True, **kwargs):
+        if value_name in kwargs:
+            return kwargs[value_name]
+        value_name_upper = value_name.upper()
+        if value_name_upper in environ:
+            return environ[value_name_upper]
+        if raise_not_found:
             raise ValueError(f"`{value_name}` is not found.")
-        return value
+        return None
 
 
 @dataclass
@@ -74,7 +96,7 @@ class Config(BasicConfig):
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
-        self.auth = (self._get_value("nc_auth_user", **kwargs), self._get_value("nc_auth_pass", **kwargs))
+        self.auth = (self._get_config_value("nc_auth_user", **kwargs), self._get_config_value("nc_auth_pass", **kwargs))
 
 
 @dataclass
@@ -92,12 +114,12 @@ class AppConfig(BasicConfig):
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
-        self.ae_version = self._get_value("ae_version", raise_not_found=False, **kwargs)
+        self.ae_version = self._get_config_value("ae_version", raise_not_found=False, **kwargs)
         if not self.ae_version:
             self.ae_version = "1.0.0"
-        self.app_name = self._get_value("app_id", **kwargs)
-        self.app_version = self._get_value("app_version", **kwargs)
-        self.app_secret = self._get_value("app_secret", **kwargs).encode("UTF-8")
+        self.app_name = self._get_config_value("app_id", **kwargs)
+        self.app_version = self._get_config_value("app_version", **kwargs)
+        self.app_secret = self._get_config_value("app_secret", **kwargs).encode("UTF-8")
 
 
 class NcSessionBasic(ABC):
@@ -147,7 +169,7 @@ def _ocs(self, method: str, path_params: str, headers: dict, data: Optional[byte
         info = f"request: method={method}, url={url_params}"
         nested_req = kwargs.pop("nested_req", False)
         try:
-            timeout = kwargs.pop("timeout", options.TIMEOUT)
+            timeout = kwargs.pop("timeout", self.cfg.options.timeout)
             if method == "GET":
                 response = self.adapter.get(url_params, headers=headers, timeout=timeout, **kwargs)
             else:
@@ -190,14 +212,15 @@ def dav_stream(
 
     def _dav(self, method: str, path: str, headers: dict, data: Optional[bytes], **kwargs) -> Response:
         self.init_adapter()
-        timeout = kwargs.pop("timeout", options.TIMEOUT_DAV)
+        # self.cfg.
+        timeout = kwargs.pop("timeout", self.cfg.options.timeout_dav)
         return self.adapter.request(
             method, self.cfg.endpoint + path, headers=headers, content=data, timeout=timeout, **kwargs
         )
 
     def _dav_stream(self, method: str, path: str, headers: dict, data: Optional[bytes], **kwargs) -> Iterator[Response]:
         self.init_adapter()
-        timeout = kwargs.pop("timeout", options.TIMEOUT_DAV)
+        timeout = kwargs.pop("timeout", self.cfg.options.timeout_dav)
         return self.adapter.stream(
             method, self.cfg.endpoint + path, headers=headers, content=data, timeout=timeout, **kwargs
         )
@@ -249,9 +272,7 @@ def __init__(self, **kwargs):
         super().__init__(user=self.cfg.auth[0])
 
     def _create_adapter(self) -> Client:
-        return Client(
-            auth=self.cfg.auth, follow_redirects=True, limits=self.limits, verify=options.VERIFY_NC_CERTIFICATE
-        )
+        return Client(auth=self.cfg.auth, follow_redirects=True, limits=self.limits, verify=self.cfg.options.nc_cert)
 
 
 class NcSessionApp(NcSessionBasic):
@@ -274,7 +295,7 @@ def _dav_stream(self, method: str, path: str, headers: dict, data: Optional[byte
         return super()._dav_stream(method, path, headers, data, **kwargs)
 
     def _create_adapter(self) -> Client:
-        adapter = Client(follow_redirects=True, limits=self.limits, verify=options.VERIFY_NC_CERTIFICATE)
+        adapter = Client(follow_redirects=True, limits=self.limits, verify=self.cfg.options.nc_cert)
         adapter.headers.update(
             {
                 "AE-VERSION": self.cfg.ae_version,

nc_py_api/options.py

@@ -1,15 +1,25 @@
-"""Options to change nc_py_api's runtime behaviour."""
+"""Options to change nc_py_api's runtime behavior.
 
-XDEBUG_SESSION = "PHPSTORM"
-"""Dev option, for debugging PHP code"""
+Each setting only affects newly created instances of the Nextcloud or NextcloudApp class, unless otherwise specified.
+Refer to the documentation for information in which different ways it can be useful.
+"""
+from os import environ
 
-TIMEOUT = 50
+from dotenv import load_dotenv
+
+load_dotenv()
+
+XDEBUG_SESSION = environ.get("XDEBUG_SESSION", "")
+"""Dev option, for debugging PHP code."""
+
+NPA_TIMEOUT = environ.get("NPA_TIMEOUT", 50)
 """Default timeout for OCS API calls. Set to "None" to disable timeouts for development."""
 
-TIMEOUT_DAV = TIMEOUT * 3 if TIMEOUT else None
+NPA_TIMEOUT_DAV = environ.get("NPA_TIMEOUT_DAV", NPA_TIMEOUT * 3 if isinstance(NPA_TIMEOUT, int) else None)
 """File operations timeout, usually it is OCS timeout multiplied by 3."""
 
-VERIFY_NC_CERTIFICATE = True
-"""Option to enable/disable Nextcloud certificate verification
+NPA_NC_CERT = environ.get("NPA_NC_CERT", True)
+"""Option to enable/disable Nextcloud certificate verification.
 
-In the case of self-signed certificates, you can disable verification."""
+SSL certificates (a.k.a CA bundle) used to  verify the identity of requested hosts. Either `True` (default CA bundle),
+a path to an SSL certificate file, or `False` (which will disable verification)."""

pyproject.toml

@@ -43,16 +43,17 @@ dynamic = [
   "version",
 ]
 dependencies = [
-  "fastapi==0.101",
-  "httpx==0.24.1",
-  "pydantic==2.1.1",
-  "requests==2.31",
-  "xmltodict==0.13",
+  "fastapi>=0.101",
+  "httpx>=0.24.1",
+  "pydantic>=2.1.1",
+  "python-dotenv>=1",
+  "requests>=2.31",
+  "xmltodict>=0.13",
 ]
 [project.optional-dependencies]
 app = [
-  "uvicorn[standard]==0.23.2",
-  "xxhash==3.3",
+  "uvicorn[standard]>=0.23.2",
+  "xxhash>=3.3",
 ]
 bench = [
   "matplotlib",

tests/misc_test.py

@@ -39,10 +39,10 @@ def test_require_capabilities():
 
 
 def test_config_get_value():
-    BasicConfig()._get_value("non_exist_value", raise_not_found=False)
+    BasicConfig()._get_config_value("non_exist_value", raise_not_found=False)
     with pytest.raises(ValueError):
-        BasicConfig()._get_value("non_exist_value")
-    assert BasicConfig()._get_value("non_exist_value", non_exist_value=123) == 123
+        BasicConfig()._get_config_value("non_exist_value")
+    assert BasicConfig()._get_config_value("non_exist_value", non_exist_value=123) == 123
 
 
 def test_deffered_error():