feat: add open-data-capture cli tool

Author: joshunrau

cli/open-data-capture

@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import sys
+
+if sys.version_info[:2] < (3, 8):
+    print('Error: Python 3.8 or higher is required', file=sys.stderr)
+    sys.exit(1)
+
+
+import json
+import os
+
+from argparse import ArgumentTypeError, ArgumentParser
+from typing import Any, Callable
+from urllib.parse import urlparse
+from urllib.request import HTTPError, Request, urlopen
+from urllib.response import addinfourl
+from urllib.error import URLError
+
+PROGRAM_NAME = os.path.basename(sys.argv[0])
+USER_CONFIG_FILEPATH = os.path.expanduser('~/.odc-cli.json')
+
+
+config: Config
+
+
+def require_url(fn: Callable[..., Any]) -> Callable[..., Any]:
+    def wrapper(*args: Any, **kwargs: Any) -> None:
+        if config.base_url is None:
+            raise RuntimeError(f'URL must be defined (hint: use {PROGRAM_NAME} config set-url)')
+        fn(*args, **kwargs)
+
+    return wrapper
+
+
+class ArgumentTypes:
+    @staticmethod
+    def valid_url_or_null(url: str) -> str | None:
+        if url == 'null':
+            return None
+        parsed = urlparse(url)
+        if not all([parsed.scheme, parsed.netloc]):
+            raise ArgumentTypeError(f'invalid url: {url}')
+        return url
+
+
+class Config:
+    _dict: dict[str, Any]
+    _filepath = os.path.expanduser('~/.odc-cli.json')
+
+    def __init__(self) -> None:
+        if not os.path.exists(self._filepath):
+            self._dict = {}
+            return None
+        with open(self._filepath, 'r') as file:
+            self._dict = json.load(file)
+
+    @property
+    def access_token(self) -> str | None:
+        return self._dict.get('access_token')
+
+    @access_token.setter
+    def access_token(self, value: str | None) -> None:
+        self._dict['access_token'] = value
+
+    @property
+    def base_url(self) -> str | None:
+        return self._dict.get('base_url')
+
+    @base_url.setter
+    def base_url(self, value: str | None) -> None:
+        self._dict['base_url'] = value
+
+    def write(self) -> None:
+        with open(self._filepath, 'w') as file:
+            json.dump(self._dict, file)
+
+
+class HttpResponse:
+    def __init__(self, data: Any, status: int | None):
+        self.data = data
+        self.status = status
+
+    def __str__(self) -> str:
+        lines = [
+            f'HTTP Status: {self.status if self.status is not None else "N/A"}',
+            f'Success: {self.ok}',
+            'Data:',
+        ]
+        try:
+            pretty_data = json.dumps(self.data, indent=2, ensure_ascii=False)
+        except (TypeError, ValueError):
+            pretty_data = str(self.data)
+        lines.append(pretty_data)
+        return '\n'.join(lines)
+
+    @property
+    def ok(self) -> bool:
+        return self.status is not None and 200 <= self.status < 300
+
+
+class HttpClient:
+    @classmethod
+    def post(cls, url: str, data: Any) -> HttpResponse:
+        headers: dict[str, str] = {'Content-Type': 'application/json'}
+        json_data = json.dumps(data).encode('utf-8')
+        req = Request(url, data=json_data, headers=headers, method='POST')
+        return cls._send(req)
+
+    @staticmethod
+    def _send(req: Request) -> HttpResponse:
+        try:
+            with urlopen(req, timeout=2) as response:
+                return HttpClient._build_response(response)
+        except HTTPError as err:
+            return HttpClient._build_response(err)
+        except URLError as err:
+            reason = err.reason
+            if isinstance(reason, OSError):
+                message = f'Network error: {reason.strerror or str(reason)}'
+            else:
+                message = f'Failed to reach the server: {reason}'
+
+            return HttpResponse(
+                data={
+                    'error': 'URLError',
+                    'message': message,
+                    'hint': 'Please check if the URL is correct and the server is reachable',
+                    'url': req.get_full_url(),
+                },
+                status=None,
+            )
+
+    @staticmethod
+    def _build_response(response: addinfourl) -> HttpResponse:
+        body = response.read()
+        try:
+            data = json.loads(body.decode(errors='replace'))
+        except json.JSONDecodeError:
+            data = {'error': 'Invalid JSON response'}
+        return HttpResponse(data=data, status=response.getcode())
+
+
+class AuthCommands:
+    @require_url
+    @staticmethod
+    def login(username: str, password: str) -> None:
+        response = HttpClient.post(f'{config.base_url}/v1/auth/login', {'username': username, 'password': password})
+        if not response.ok:
+            print(response)
+            return None
+        config.access_token = response.data['accessToken']
+        print('Success!')
+
+
+class ConfigCommands:
+    @staticmethod
+    def get_url() -> None:
+        print(config.base_url or 'null')
+
+    @staticmethod
+    def set_url(url: str) -> None:
+        config.base_url = url
+        config.write()
+
+
+def main() -> None:
+    global config
+    config = Config()
+
+    parser = ArgumentParser(prog=PROGRAM_NAME)
+    subparsers = parser.add_subparsers(help='subcommand help', required=True)
+
+    ## AUTH
+    auth_parser = subparsers.add_parser('auth')
+    auth_subparsers = auth_parser.add_subparsers(required=True)
+
+    login_parser = auth_subparsers.add_parser('login')
+    login_parser.add_argument('--username', type=str, required=True)
+    login_parser.add_argument('--password', type=str, required=True)
+    login_parser.set_defaults(fn=AuthCommands.login)
+
+    ## CONFIG
+
+    config_parser = subparsers.add_parser('config')
+    config_subparsers = config_parser.add_subparsers(required=True)
+
+    get_config_url_parser = config_subparsers.add_parser('get-url')
+    get_config_url_parser.set_defaults(fn=ConfigCommands.get_url)
+
+    set_config_parser = config_subparsers.add_parser('set-url')
+    set_config_parser.add_argument('url', type=ArgumentTypes.valid_url_or_null)
+    set_config_parser.set_defaults(fn=ConfigCommands.set_url)
+
+    kwargs = vars(parser.parse_args())
+    fn = kwargs.pop('fn')
+    fn(**kwargs)
+
+
+if __name__ == '__main__':
+    main()