chore: apply pre-commit hook on all files

docs/source/conf.py

@@ -12,24 +12,23 @@
 #
 import os
 import sys
-sys.path.insert(0, os.path.abspath('..'))
+
+sys.path.insert(0, os.path.abspath(".."))
 
 
 # -- Project information -----------------------------------------------------
 
-project = 'fathomnet-py'
-copyright = '2022-2024, Monterey Bay Aquarium Research Institute'
-author = 'Kevin Barnard'
+project = "fathomnet-py"
+copyright = "2022-2024, Monterey Bay Aquarium Research Institute"
+author = "Kevin Barnard"
 
 
 # -- General configuration ---------------------------------------------------
 
 # Add any Sphinx extension module names here, as strings. They can be
 # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
 # ones.
-extensions = [
-    'sphinx.ext.autodoc'
-]
+extensions = ["sphinx.ext.autodoc"]
 
 # Add any paths that contain templates here, relative to this directory.
 templates_path = []
@@ -45,7 +44,7 @@
 # The theme to use for HTML and HTML Help pages.  See the documentation for
 # a list of builtin themes.
 #
-html_theme = 'sphinx_rtd_theme'
+html_theme = "sphinx_rtd_theme"
 
 # Add any paths that contain custom static files (such as style sheets) here,
 # relative to this directory. They are copied after the builtin static files,

fathomnet/api/__init__.py

@@ -9,42 +9,62 @@
 
 
 class EndpointManager:
-    ROOT = 'http://fathomnet.org:8080'
+    ROOT = "http://fathomnet.org:8080"
     PATH = None
 
     @classmethod
     def url(cls, endpoint: str) -> str:
         if cls.PATH is None:
             raise NotImplementedError
-        return '/'.join([cls.ROOT, cls.PATH, endpoint])
+        return "/".join([cls.ROOT, cls.PATH, endpoint])
 
     @classmethod
-    def request(cls, method: str, endpoint: str, parse_json: bool = True, **kwargs) -> Union[requests.Response, dict, list]:
+    def request(
+        cls, method: str, endpoint: str, parse_json: bool = True, **kwargs
+    ) -> Union[requests.Response, dict, list]:
         url = cls.url(endpoint)
         res = SESSION.request(method, url, **kwargs)
         if res.ok:  # Status code < 400
             return res.json() if parse_json else res
         elif res.status_code == 401:  # Not authorized, need to authenticate
-            raise ValueError('Unauthorized: please authenticate first.')
-        elif res.status_code == 403:  # Forbidden, can't access this endpoint with given authentication
-            raise ValueError('Forbidden: you cannot access this resource.')
+            raise ValueError("Unauthorized: please authenticate first.")
+        elif (
+            res.status_code == 403
+        ):  # Forbidden, can't access this endpoint with given authentication
+            raise ValueError("Forbidden: you cannot access this resource.")
         elif res.status_code < 500:  # User error
-            raise ValueError('Client error ({}), please check your usage (docs: https://fathomnet-py.rtfd.io/) or open an issue at https://github.com/fathomnet/fathomnet-py/issues/new with the details below.\n{}'.format(res.status_code, debug_format_response(res)))
+            raise ValueError(
+                "Client error ({}), please check your usage (docs: https://fathomnet-py.rtfd.io/) or open an issue at https://github.com/fathomnet/fathomnet-py/issues/new with the details below.\n{}".format(
+                    res.status_code, debug_format_response(res)
+                )
+            )
         else:  # Server error, debug the response
-            raise ValueError('Server error ({}), please contact the FathomNet administrators with the details below.\n\n{}'.format(res.status_code, debug_format_response(res)))
+            raise ValueError(
+                "Server error ({}), please contact the FathomNet administrators with the details below.\n\n{}".format(
+                    res.status_code, debug_format_response(res)
+                )
+            )
 
     @classmethod
-    def get(cls, endpoint: str, parse_json: bool = True, **kwargs) -> Union[requests.Response, dict, list]:
-        return cls.request('GET', endpoint, parse_json=parse_json, **kwargs)
+    def get(
+        cls, endpoint: str, parse_json: bool = True, **kwargs
+    ) -> Union[requests.Response, dict, list]:
+        return cls.request("GET", endpoint, parse_json=parse_json, **kwargs)
 
     @classmethod
-    def put(cls, endpoint: str, parse_json: bool = True, **kwargs) -> Union[requests.Response, dict, list]:
-        return cls.request('PUT', endpoint, parse_json=parse_json, **kwargs)
+    def put(
+        cls, endpoint: str, parse_json: bool = True, **kwargs
+    ) -> Union[requests.Response, dict, list]:
+        return cls.request("PUT", endpoint, parse_json=parse_json, **kwargs)
 
     @classmethod
-    def post(cls, endpoint: str, parse_json: bool = True, **kwargs) -> Union[requests.Response, dict, list]:
-        return cls.request('POST', endpoint, parse_json=parse_json, **kwargs)
+    def post(
+        cls, endpoint: str, parse_json: bool = True, **kwargs
+    ) -> Union[requests.Response, dict, list]:
+        return cls.request("POST", endpoint, parse_json=parse_json, **kwargs)
 
     @classmethod
-    def delete(cls, endpoint: str, parse_json: bool = True, **kwargs) -> Union[requests.Response, dict, list]:
-        return cls.request('DELETE', endpoint, parse_json=parse_json, **kwargs)
+    def delete(
+        cls, endpoint: str, parse_json: bool = True, **kwargs
+    ) -> Union[requests.Response, dict, list]:
+        return cls.request("DELETE", endpoint, parse_json=parse_json, **kwargs)

fathomnet/api/activity.py

@@ -2,65 +2,81 @@
 
 from typing import List, Optional
 
-from fathomnet.api import EndpointManager
 from fathomnet import dto
+from fathomnet.api import EndpointManager
 
 
 class Activity(EndpointManager):
-    PATH = 'activity'
+    PATH = "activity"
 
 
-def find_all(auth_header: Optional[dto.AuthHeader] = None,
-             start_timestamp: Optional[str] = None, end_timestamp: Optional[str] = None,
-             limit: Optional[int] = None, offset: Optional[int] = None,
-             include_self: Optional[bool] = None) -> List[dto.Activity]:
+def find_all(
+    auth_header: Optional[dto.AuthHeader] = None,
+    start_timestamp: Optional[str] = None,
+    end_timestamp: Optional[str] = None,
+    limit: Optional[int] = None,
+    offset: Optional[int] = None,
+    include_self: Optional[bool] = None,
+) -> List[dto.Activity]:
     """Get a list of all activity."""
     params = {}
     if start_timestamp:
-        params['startTimestamp'] = start_timestamp
+        params["startTimestamp"] = start_timestamp
     if end_timestamp:
-        params['endTimestamp'] = end_timestamp
+        params["endTimestamp"] = end_timestamp
     if limit:
-        params['limit'] = limit
+        params["limit"] = limit
     if offset:
-        params['offset'] = offset
+        params["offset"] = offset
     if include_self:
-        params['includeSelf'] = include_self
-    res_json = Activity.get('', params=params, auth=auth_header)
+        params["includeSelf"] = include_self
+    res_json = Activity.get("", params=params, auth=auth_header)
     return list(map(dto.Activity.from_dict, res_json))
 
 
-def find_by_email(email: str,
-                  auth_header: Optional[dto.AuthHeader] = None,
-                  start_timestamp: Optional[str] = None, end_timestamp: Optional[str] = None,
-                  limit: Optional[int] = None, offset: Optional[int] = None) -> List[dto.Activity]:
+def find_by_email(
+    email: str,
+    auth_header: Optional[dto.AuthHeader] = None,
+    start_timestamp: Optional[str] = None,
+    end_timestamp: Optional[str] = None,
+    limit: Optional[int] = None,
+    offset: Optional[int] = None,
+) -> List[dto.Activity]:
     """Get a list of activity by email."""
     params = {}
     if start_timestamp:
-        params['startTimestamp'] = start_timestamp
+        params["startTimestamp"] = start_timestamp
     if end_timestamp:
-        params['endTimestamp'] = end_timestamp
+        params["endTimestamp"] = end_timestamp
     if limit:
-        params['limit'] = limit
+        params["limit"] = limit
     if offset:
-        params['offset'] = offset
-    res_json = Activity.get('query/email/{}'.format(email), params=params, auth=auth_header)
+        params["offset"] = offset
+    res_json = Activity.get(
+        "query/email/{}".format(email), params=params, auth=auth_header
+    )
     return list(map(dto.Activity.from_dict, res_json))
 
 
-def find_by_email_admin(email: str,
-                        auth_header: Optional[dto.AuthHeader] = None,
-                        start_timestamp: Optional[str] = None, end_timestamp: Optional[str] = None,
-                        limit: Optional[int] = None, offset: Optional[int] = None) -> List[dto.Activity]:
+def find_by_email_admin(
+    email: str,
+    auth_header: Optional[dto.AuthHeader] = None,
+    start_timestamp: Optional[str] = None,
+    end_timestamp: Optional[str] = None,
+    limit: Optional[int] = None,
+    offset: Optional[int] = None,
+) -> List[dto.Activity]:
     """(Admin) Get a list of activity by email. Used to support notification applications."""
     params = {}
     if start_timestamp:
-        params['startTimestamp'] = start_timestamp
+        params["startTimestamp"] = start_timestamp
     if end_timestamp:
-        params['endTimestamp'] = end_timestamp
+        params["endTimestamp"] = end_timestamp
     if limit:
-        params['limit'] = limit
+        params["limit"] = limit
     if offset:
-        params['offset'] = offset
-    res_json = Activity.get('admin/query/email/{}'.format(email), params=params, auth=auth_header)
+        params["offset"] = offset
+    res_json = Activity.get(
+        "admin/query/email/{}".format(email), params=params, auth=auth_header
+    )
     return list(map(dto.Activity.from_dict, res_json))