From d547ecc441793d4498a763bf28d145ddc0934b53 Mon Sep 17 00:00:00 2001 From: Justaus3r <72691864+Justaus3r@users.noreply.github.com> Date: Wed, 1 Mar 2023 17:03:25 +0000 Subject: [PATCH] release: bump release to 0.5.0 --- ceg/__init__.py | 140 +- ceg/api.py | 434 +++--- ceg/arg_parser.py | 292 ++-- ceg/ceg.py | 1190 ++++++++-------- ceg/cli.py | 164 +-- ceg/exceptions.py | 164 +-- ceg/logger.py | 152 +-- ceg/misc.py | 418 +++--- docs/ceg.html | 39 +- docs/ceg/api.html | 55 +- docs/ceg/arg_parser.html | 318 ++--- docs/ceg/ceg.html | 2754 ++++++++++++++++++++------------------ docs/ceg/cli.html | 187 +-- docs/ceg/exceptions.html | 89 +- docs/ceg/logger.html | 316 ++--- docs/ceg/misc.html | 205 +-- docs/search.js | 89 +- pyproject.toml | 3 +- 18 files changed, 3506 insertions(+), 3503 deletions(-) mode change 100755 => 100644 ceg/__init__.py mode change 100755 => 100644 ceg/api.py mode change 100755 => 100644 ceg/arg_parser.py mode change 100755 => 100644 ceg/ceg.py mode change 100755 => 100644 ceg/cli.py mode change 100755 => 100644 ceg/exceptions.py mode change 100755 => 100644 ceg/logger.py mode change 100755 => 100644 ceg/misc.py diff --git a/ceg/__init__.py b/ceg/__init__.py old mode 100755 new mode 100644 index 015b0ef..0363d19 --- a/ceg/__init__.py +++ b/ceg/__init__.py @@ -1,70 +1,70 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" -Introduction -======== -Ceg (as in **c**r**e**ate **g**ist and pronounced *Keg*) is a command-line utility as well as a library for -interacting with github gists.it uses github's official api for performing all operations.it can: -- Create gists. -- Modify existing gists. -- Download gists. -- List public/private(secret) gists for authenticated users as well as list public gists for unauthenticated users. -- Delete a gist. -- Create local backup of all the gists. - -Installation -============ -There are multiple ways to install ceg,the simplest one being installing from PYPI: -``` -# py instead of python3 on windows -python3 -m pip install ceg -``` -You can also install it manually.for that you need to have [``poetry``](https://python-poetry.org/docs/master/#installing-with-the-official-installer) installed and be on a system with minimal python version being 3.7.after installing poetry,you can just -do `poetry build` and pip install from `dist/ceg*.whl` or whatever you prefer.please be mindful that installing poetry -from pip is [not recommended](https://python-poetry.org/docs/#alternative-installation-methods-not-recommended). -``` -# you can also use install/uninstall scripts after cloning the repo, if on *nix. -curl -sSL https://install.python-poetry.org | python3 - -git clone https://github.com/justaus3r/ceg.git -cd ceg -poetry build -``` - -Now wat? -======= -After installing ceg you can either do ``ceg --help`` in your terminal, check out projects README or refer to api documentation. - -**Note:** -Please only refer to submodule named `api` as other files contain reference to main implementation of ceg and such -may contain ambiguous documentation which may or maynot be clear unless codebase is understood. - -""" - -from .misc import UtilInfo -from .api import CegApi - - -__author__ = "Justaus3r" -__email__ = "x-neron@pm.me" -__version__ = UtilInfo.VERSION -__description__ = UtilInfo.DESCRIPTION +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" +Introduction +======== +Ceg (as in **c**r**e**ate **g**ist and pronounced *Keg*) is a command-line utility as well as a library for +interacting with github gists.it uses github's official api for performing all operations.it can: +- Create gists. +- Modify existing gists. +- Download gists. +- List public/private(secret) gists for authenticated users as well as list public gists for unauthenticated users. +- Delete a gist. +- Create local backup of all the gists. + +Installation +============ +There are multiple ways to install ceg,the simplest one being installing from PYPI: +``` +# py instead of python3 on windows +python3 -m pip install ceg +``` +You can also install it manually.for that you need to have [``poetry``](https://python-poetry.org/docs/master/#installing-with-the-official-installer) installed and be on a system with minimal python version being 3.7.after installing poetry,you can just +do `poetry build` and pip install from `dist/ceg*.whl` or whatever you prefer.please be mindful that installing poetry +from pip is [not recommended](https://python-poetry.org/docs/#alternative-installation-methods-not-recommended). +``` +# you can also use install/uninstall scripts after cloning the repo, if on *nix. +curl -sSL https://install.python-poetry.org | python3 - +git clone https://github.com/justaus3r/ceg.git +cd ceg +poetry build +``` + +Now wat? +======= +After installing ceg you can either do ``ceg --help`` in your terminal, check out projects README or refer to api documentation. + +**Note:** +Please only refer to submodule named `api` as other files contain reference to main implementation of ceg and such +may contain ambiguous documentation which may or maynot be clear unless codebase is understood. + +""" + +from .misc import UtilInfo +from .api import CegApi + + +__author__ = "Justaus3r" +__email__ = "x-neron@pm.me" +__version__ = UtilInfo.VERSION +__description__ = UtilInfo.DESCRIPTION diff --git a/ceg/api.py b/ceg/api.py old mode 100755 new mode 100644 index c5cf187..bf6f39a --- a/ceg/api.py +++ b/ceg/api.py @@ -1,217 +1,217 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" -Description ------------ - -This is the main api of ceg.it internally uses the main implementation of ceg.interface provided to a developer -is a single class named CegApi, which contains all the functionality of ceg.it can be instantiated by providing -github secret key, which can be obtained from developers **``github``** >> **``Settings``** >> **``Developer settings``** >> **``Personal access tokens``**. - -Typical usage example ---------------------- -Object instantiation: -``` -cgi = CegApi(secret_key="abcd") -``` -For creating a gist: -``` -gist_url = cgi.post("file1.py", "file2.py", "dirty_secrets.verysecurefile", is_private=False, gist_description="bla") -# note that if a file provided as argument does not exist,ceg will automatically create it and open -# it in your default file editor. -``` - -For modifying an existing gist: -``` -response_str = cgi.patch("file2.py", gist_id="abc") -# you can optionally also modify the gist description using 'gist_description="whatever"' -``` - -For listing gists for authenticated user: -``` -user_gist_list = cgi.list() -``` - -For list gist for unauthenticated user: -``` -user_gist_list = cgi.list_other("username") -``` - -For downloading a gist: -``` -response_str = cgi.get("gistid1", "gistid2") -# note that gist-ids are typically hashes like 'aa5a315d61ae9438b18d' -``` - -For creating a backup: -``` -response_str = cgi.backup() -``` - -For deleting a gist: -``` -cgi.delete("gistid") -# doesn't support batch operation for now -``` - -Api Reference -------------- -""" - -from .ceg import Ceg -from typing import List, Dict, Optional - - -class CegApi: - """Main interface for api. - - Provides the main interface open for api.contains all - the methods needed to perform all basic operations on gists. - - Attributes: - ceg_instance: its an instance of Ceg class.which contains the main - implementation of ceg utility. - """ - - def __init__(self, secret_key: str) -> None: - """Inits CegApi with github secret key""" - self.ceg_instance: Ceg = Ceg( - operation="", - arg_value="", - is_recursive_operation=False, - is_other_user=False, - secret_key=secret_key, - do_logging=False, - gist_no_public=False, - gist_desc="", - gist_id="", - ) - - def get(self, *args: str) -> str: - """Downloads a gist. - - Receives arbitrary amount gist-ids and downloads them. - - Args: - *args: Variable lenght argument list containing gist-ids. - - - Returns: - Returns HTTP call response status in string format. - """ - self.ceg_instance.http_operation = "get" - self.ceg_instance.arg_val = args - self.ceg_instance.get() - return self.ceg_instance.response_status_str - - def post( - self, - *args: str, - is_private: bool = False, - gist_description: Optional[str] = None - ) -> str: - """Create arbitrary number of gists. - - Args: - *args: variable lenght arguments list containing gist-ids. - is_private: indicates whether to make gist private. - gist_description: Description for the gist. - - Returns: - Returns HTML url for newly created gist. - """ - self.ceg_instance.http_operation = "post" - self.ceg_instance.arg_val = args - self.ceg_instance.gist_no_public = is_private - self.ceg_instance.gist_description = gist_description - # type casting because of distinct variable types(i.e Optional[str] and str) - # and so so mypy will complain if not type casted - gist_html_url: str = str(self.ceg_instance.post()) - return gist_html_url - - def patch( - self, *args: str, gist_id: str, gist_description: Optional[str] = None - ) -> str: - """Modify arbitrary number of existing gists. - - Args: - *args: variable lenght arguments list containing gist-names. - gist_id: gist-id for the gist,that is to be modified. - gist_description: (Optional) Description for the gist. - - Returns: - Returns HTTP call response status in string format. - """ - self.ceg_instance.http_operation = "patch" - self.ceg_instance.arg_val = list(args) # type: ignore - self.ceg_instance.gist_description = gist_description - self.ceg_instance.gist_id = gist_id - self.ceg_instance.patch() - return self.ceg_instance.response_status_str - - def delete(self, *args) -> str: - """Delete an existing gist. - - Args: - gist_id = arbitrary amount of gist-ids. - - Returns: - Returns HTTP call response status in string format. - """ - self.ceg_instance.http_operation = "delete" - self.ceg_instance.arg_val = args - self.ceg_instance.delete() - return self.ceg_instance.response_status_str - - def list(self) -> Optional[List[Dict[str, str]]]: - """Return gist data for authenticated user. - - Returns: - Returns a sequence containing list of all the gists of user with some info. - """ - self.ceg_instance.http_operation = "get" - return self.ceg_instance.list() - - def list_other(self, user_name: str) -> Optional[List[Dict[str, str]]]: - """Return gist data for unauthenticated user. - - Args: - user_name: username for the user. - - Returns: - Returns a sequence containing list of all the gists of user with some info. - """ - self.ceg_instance.http_operation = "get" - self.ceg_instance.arg_val = user_name - return self.ceg_instance.list_other() - - def backup(self) -> str: - """Create backup of all gists on local media. - - Returns: - Returns HTTP call response status in string format. - """ - self.ceg_instance.http_operation = "get" - self.ceg_instance.is_recursive_op = True - self.ceg_instance.backup() - return self.ceg_instance.response_status_str +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" +Description +----------- + +This is the main api of ceg.it internally uses the main implementation of ceg.interface provided to a developer +is a single class named CegApi, which contains all the functionality of ceg.it can be instantiated by providing +github secret key, which can be obtained from developers **``github``** >> **``Settings``** >> **``Developer settings``** >> **``Personal access tokens``**. + +Typical usage example +--------------------- +Object instantiation: +``` +cgi = CegApi(secret_key="abcd") +``` +For creating a gist: +``` +gist_url = cgi.post("file1.py", "file2.py", "dirty_secrets.verysecurefile", is_private=False, gist_description="bla") +# note that if a file provided as argument does not exist,ceg will automatically create it and open +# it in your default file editor. +``` + +For modifying an existing gist: +``` +response_str = cgi.patch("file2.py", gist_id="abc") +# you can optionally also modify the gist description using 'gist_description="whatever"' +``` + +For listing gists for authenticated user: +``` +user_gist_list = cgi.list() +``` + +For list gist for unauthenticated user: +``` +user_gist_list = cgi.list_other("username") +``` + +For downloading a gist: +``` +response_str = cgi.get("gistid1", "gistid2") +# note that gist-ids are typically hashes like 'aa5a315d61ae9438b18d' +``` + +For creating a backup: +``` +response_str = cgi.backup() +``` + +For deleting a gist: +``` +cgi.delete("gistid") +# doesn't support batch operation for now +``` + +Api Reference +------------- +""" + +from .ceg import Ceg +from typing import List, Dict, Optional + + +class CegApi: + """Main interface for api. + + Provides the main interface open for api.contains all + the methods needed to perform all basic operations on gists. + + Attributes: + ceg_instance: its an instance of Ceg class.which contains the main + implementation of ceg utility. + """ + + def __init__(self, secret_key: str) -> None: + """Inits CegApi with github secret key""" + self.ceg_instance: Ceg = Ceg( + operation="", + arg_value="", + is_recursive_operation=False, + is_other_user=False, + secret_key=secret_key, + do_logging=False, + gist_no_public=False, + gist_desc="", + gist_id="", + ) + + def get(self, *args: str) -> str: + """Downloads a gist. + + Receives arbitrary amount gist-ids and downloads them. + + Args: + *args: Variable lenght argument list containing gist-ids. + + + Returns: + Returns HTTP call response status in string format. + """ + self.ceg_instance.http_operation = "get" + self.ceg_instance.arg_val = args + self.ceg_instance.get() + return self.ceg_instance.response_status_str + + def post( + self, + *args: str, + is_private: bool = False, + gist_description: Optional[str] = None + ) -> str: + """Create arbitrary number of gists. + + Args: + *args: variable lenght arguments list containing gist-ids. + is_private: indicates whether to make gist private. + gist_description: Description for the gist. + + Returns: + Returns HTML url for newly created gist. + """ + self.ceg_instance.http_operation = "post" + self.ceg_instance.arg_val = args + self.ceg_instance.gist_no_public = is_private + self.ceg_instance.gist_description = gist_description + # type casting because of distinct variable types(i.e Optional[str] and str) + # and so so mypy will complain if not type casted + gist_html_url: str = str(self.ceg_instance.post()) + return gist_html_url + + def patch( + self, *args: str, gist_id: str, gist_description: Optional[str] = None + ) -> str: + """Modify arbitrary number of existing gists. + + Args: + *args: variable lenght arguments list containing gist-names. + gist_id: gist-id for the gist,that is to be modified. + gist_description: (Optional) Description for the gist. + + Returns: + Returns HTTP call response status in string format. + """ + self.ceg_instance.http_operation = "patch" + self.ceg_instance.arg_val = list(args) # type: ignore + self.ceg_instance.gist_description = gist_description + self.ceg_instance.gist_id = gist_id + self.ceg_instance.patch() + return self.ceg_instance.response_status_str + + def delete(self, *args) -> str: + """Delete an existing gist. + + Args: + gist_id = arbitrary amount of gist-ids. + + Returns: + Returns HTTP call response status in string format. + """ + self.ceg_instance.http_operation = "delete" + self.ceg_instance.arg_val = args + self.ceg_instance.delete() + return self.ceg_instance.response_status_str + + def list(self) -> Optional[List[Dict[str, str]]]: + """Return gist data for authenticated user. + + Returns: + Returns a sequence containing list of all the gists of user with some info. + """ + self.ceg_instance.http_operation = "get" + return self.ceg_instance.list() + + def list_other(self, user_name: str) -> Optional[List[Dict[str, str]]]: + """Return gist data for unauthenticated user. + + Args: + user_name: username for the user. + + Returns: + Returns a sequence containing list of all the gists of user with some info. + """ + self.ceg_instance.http_operation = "get" + self.ceg_instance.arg_val = user_name + return self.ceg_instance.list_other() + + def backup(self) -> str: + """Create backup of all gists on local media. + + Returns: + Returns HTTP call response status in string format. + """ + self.ceg_instance.http_operation = "get" + self.ceg_instance.is_recursive_op = True + self.ceg_instance.backup() + return self.ceg_instance.response_status_str diff --git a/ceg/arg_parser.py b/ceg/arg_parser.py old mode 100755 new mode 100644 index be069a6..6fd4a93 --- a/ceg/arg_parser.py +++ b/ceg/arg_parser.py @@ -1,145 +1,147 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" Argument parser for ceg """ - -import sys -import argparse -from .misc import UtilInfo - - -class ArgumentParser(argparse.ArgumentParser): - """Reccord arguments from cli. - - Argument parser that inherits from argparse.ArgumentParser - class and is used for reccording all the arguments from cli. - """ - - def __init__(self) -> None: - """Inits (Parent) ArgumentParser with program name and description""" - super().__init__( - prog=UtilInfo.UTIL_NAME, - formatter_class=argparse.RawDescriptionHelpFormatter, - usage=UtilInfo.UTIL_USAGE, - description=UtilInfo.DESCRIPTION, - epilog=UtilInfo.EPILOG, - ) - - def reccord_arguments(self) -> argparse.Namespace: - """Record (possibly) conflicting arguments from commandline. - - arguments that conflict with each other,i.e: not meant to be used - simultaneously will be reccorded with add_mutually_exclusive_group() - method while normal arguments will be reccorded with add_argument() - method. - - Returns: - Returns the argparse.Namespace object containing all the arguments. - - """ - # TODO: for [v0.2.0 - v1.0.0]: switch for returning enhanced return codes for better script compatibility. - # TODO: for [v0.2.0 - v1.0.0]: Use pickle to use serialized cache from local storage(Security implications?). - # TODO: for [v0.2.0 - v1.0.0]: Change arg parser to flask/click or python-poetry/cleo - group = self.add_mutually_exclusive_group() - group.add_argument( - "-po", - "--post", - help="create a gist", - metavar="GISTNAME", - type=str, - nargs="+", - ) - # anonymous auxiliary arguments for --post(some also maybe used for --patch). - self.add_argument( - "-np", "--no-public", action="store_true", help=argparse.SUPPRESS - ) - self.add_argument("-desc", "--description", type=str, help=argparse.SUPPRESS) - - group.add_argument( - "-pa", - "--patch", - help="modify an existing gist", - metavar="GISTNAME", - type=str, - nargs="+", - ) - # anonymous auxiliary arguments for --patch - self.add_argument("-gi", "--gist-id", type=str, help=argparse.SUPPRESS) - - group.add_argument( - "-g", - "--get", - help="download gist(s)", - metavar="GISTID", - type=str, - nargs="+", - ) - group.add_argument( - "-d", - "--delete", - help="remove gist(s)", - metavar="GISTID", - type=str, - nargs="+", - ) - group.add_argument( - "-l", - "--list", - help="list public/private gists for authenticated user", - action="store_true", - ) - group.add_argument( - "-lo", - "--list-other", - help="list public gists for unauthenticated users", - metavar="USERNAME", - type=str, - ) - group.add_argument( - "-bk", "--backup", help="create a backup of all gists", action="store_true" - ) - self.add_argument( - "-sk", - "--secret-key", - help="user's github secret key", - metavar="SECRETKEY", - type=str, - ) - self.add_argument( - "-nl", - "--no-logging", - help="don't log anything to stdout", - action="store_true", - ) - - self.add_argument( - "-v", - "--version", - help="show utility's semantic version", - action="version", - version=f"{UtilInfo.UTIL_NAME} version: {UtilInfo.VERSION}", - ) - - if len(sys.argv) < 2: - self.print_help() - sys.exit(1) - return self.parse_args() +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" Argument parser for ceg """ + +import sys +import argparse +from .misc import UtilInfo + + +class ArgumentParser(argparse.ArgumentParser): + """Reccord arguments from cli. + + Argument parser that inherits from argparse.ArgumentParser + class and is used for reccording all the arguments from cli. + """ + + def __init__(self) -> None: + """Inits (Parent) ArgumentParser with program name and description""" + super().__init__( + prog=UtilInfo.UTIL_NAME, + formatter_class=argparse.RawDescriptionHelpFormatter, + usage=UtilInfo.UTIL_USAGE, + description=UtilInfo.DESCRIPTION, + epilog=UtilInfo.EPILOG, + ) + + def reccord_arguments(self) -> argparse.Namespace: + """Record (possibly) conflicting arguments from commandline. + + arguments that conflict with each other,i.e: not meant to be used + simultaneously will be reccorded with add_mutually_exclusive_group() + method while normal arguments will be reccorded with add_argument() + method. + + Returns: + Returns the argparse.Namespace object containing all the arguments. + + """ + # TODO: for [v0.2.0 - v1.0.0]: switch for returning enhanced return codes for better script compatibility. + # TODO: for [v0.2.0 - v1.0.0]: Use pickle to use serialized cache from local storage(Security implications?). + # TODO: for [v0.2.0 - v1.0.0]: Change arg parser to flask/click or python-poetry/cleo + group = self.add_mutually_exclusive_group() + group.add_argument( + "-po", + "--post", + help="create a gist", + metavar="GISTNAME", + type=str, + nargs="+", + ) + # anonymous auxiliary arguments for --post(some also maybe used for --patch). + self.add_argument( + "-np", "--no-public", action="store_true", help=argparse.SUPPRESS + ) + self.add_argument("-desc", "--description", type=str, help=argparse.SUPPRESS) + + group.add_argument( + "-pa", + "--patch", + help="modify an existing gist", + metavar="GISTNAME", + type=str, + nargs="+", + ) + # anonymous auxiliary arguments for --patch + self.add_argument("-gi", "--gist-id", type=str, help=argparse.SUPPRESS) + + group.add_argument( + "-g", + "--get", + help="download gist(s)", + metavar="GISTID", + type=str, + nargs="+", + ) + group.add_argument( + "-d", + "--delete", + help="remove gist(s)", + metavar="GISTID", + type=str, + nargs="+", + ) + group.add_argument( + "-l", + "--list", + help="list public/private gists for a user", + metavar="OPT-USERNAME", + type=str, + nargs="?", + const="self", + ) + group.add_argument( + "-bk", + "--backup", + help="create a backup of all gists", + metavar="OPT-USERNAME", + type=str, + nargs="?", + const="self", + ) + self.add_argument( + "-sk", + "--secret-key", + help="user's github secret key", + metavar="SECRETKEY", + type=str, + ) + self.add_argument( + "-nl", + "--no-logging", + help="don't log anything to stdout", + action="store_true", + ) + + self.add_argument( + "-v", + "--version", + help="show utility's semantic version", + action="version", + version=f"{UtilInfo.UTIL_NAME} version: {UtilInfo.VERSION}", + ) + + if len(sys.argv) < 2: + self.print_help() + sys.exit(1) + return self.parse_args() diff --git a/ceg/ceg.py b/ceg/ceg.py old mode 100755 new mode 100644 index d458fc6..74d3967 --- a/ceg/ceg.py +++ b/ceg/ceg.py @@ -1,571 +1,619 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" Main implementation of ceg """ - -import os -import json -import requests -from .logger import Logger -from rich.tree import Tree -from rich.console import Console -from .exceptions import GenericReturnCodes, CegExceptions -from .misc import Misc, FileHandler, open_file, gist_filename_validated -from typing import List, Tuple, Dict, Optional, Union, Callable - -console: Console = Console() - -__all__ = ("Ceg",) - - -class AuxSequence(List): - """An Auxiliary sequence. - - An Auxiliary Sequence that structurally saves stdout for - delivering api. - - Attributes: - single_item_dict: A dictionary containing elements of single gist. - """ - - def __init__(self) -> None: - self.single_item_dict: Dict[str, Union[str, List[str]]] = {} - - def append( - self, - writable: Optional[Union[str, Tree, Dict[str, List[str]]]], - iteration_complete: bool = False, - ) -> None: - """Overloads the append method of List - - Receives (unsanitized) data which is added to - a mapping,which is then appended to the - sequence itself. - - Args: - writable: Actual stream which is to be stored. - iteration_complete: A boolean acting as a delimiter for gists. - """ - if isinstance(writable, Dict): - self.single_item_dict.update(writable) - else: - key, val = writable.split(":") # type: ignore - self.single_item_dict.update({key: val}) - if iteration_complete: - single_item_dict_copy = self.single_item_dict.copy() - super().append(single_item_dict_copy) - self.single_item_dict.clear() - - -class WriteStdout: - """Write stdout to an initialized resource. - - Implements the "write stdout" functionality.It writes the stdout to an initialized resource. - - Attributes: - to_stdout: A bool indicating whether to write to stdout or not. - stream_cache: A list containing all gists info. - write_stdout: A Callable which either writes to stdout or stream_cache. - """ - - def __init__(self, to_stdout: bool) -> None: - """Inits WriteStdout with appropriate attributes""" - self.to_stdout: bool = to_stdout - self.stream_cache: AuxSequence = AuxSequence() - self.write_stdout: Callable[ - [Optional[Union[str, Tree, Dict[str, List[str]]]], bool], None - ] - # mypy was being bitchy when using ternary operator so ... - # related issue: https://github.com/python/mypy/issues/10740 - if self.to_stdout: - self.write_stdout = lambda writable, _: console.print(writable) - else: - self.write_stdout = self.stream_cache.append - - def __call__( - self, - writable: Optional[Union[str, Tree, Dict[str, List[str]]]] = None, - is_rule: bool = False, - silent: bool = False, - text_style: Optional[str] = None, - iteration_complete: bool = False, - ) -> None: - """Implements dunder call - - Calls the actual write_stdout method after evaluation. - - Args: - writable: A string or Tree object. - is_rule: A boolean indicating if the writable is a rule object. - is_tree: A boolean indicating if the writable is a Tree object. - silent: A boolean indicating whether to waste the write_stdout - call(mostly while self.to_stdout is False and we don't - want to print the rule). - text_style: Contains the styling string. - - """ - if isinstance(writable, Tree) and not self.to_stdout: - tree_dict: Dict[str, List[str]] = {"Files": []} - retreive_file: Callable[ - [Tree], str - ] = lambda tree_obj: tree_obj.label # type: ignore - tree_files: List[str] = list(map(retreive_file, writable.children)) - tree_dict["Files"] = tree_files - writable = tree_dict - if not silent: - writable = text_style + writable if text_style and self.to_stdout else writable # type: ignore - self.write_stdout( - writable, iteration_complete - ) if not is_rule else console.rule( - writable # type: ignore - ) - - -class Ceg: - """Main implementation of ceg. - - The Ceg class actually contains the main implementation - of the utility. - - Attributes: - http_operation: A Callable that actually performs the http calls. - arg_val: A string that is used as argument value as well as a switch for operation resolution. - is_recursive_op: A boolean indicating if the operation is recursive. - header: A Dict containing the HTTP header. - end_point: Endpoint for api calls. - payload: payload containing data for post requests. - to_stdout: boolean indicating whether to send data to stdout. - gist_no_public: boolean indicating if a gist is private. - is_other: boolean indicating if to list gists for unauth user. - gist_description: String containing gist description which can be used in patch(),post(). - gist_id: String containing gist-id. - response_status_str: Contains HTTP call response status in string format. - logger: Logger object. - """ - - def __init__( - self, - operation: str, - arg_value: Optional[Union[str, Tuple[str, ...]]], - is_recursive_operation: bool, - is_other_user: bool, - secret_key: Optional[str], - do_logging: bool, - gist_no_public: bool, - gist_desc: Optional[str], - gist_id: Optional[str], - ) -> None: - """Inits Ceg with appropriate attributes""" - self.http_operation: str = operation - self.arg_val: Optional[Union[str, Tuple[str, ...]]] = arg_value - self.to_stdout: bool = do_logging - self.is_recursive_op: bool = is_recursive_operation - self.is_other: bool = is_other_user - self.gist_no_public: bool = gist_no_public - self.gist_description: Optional[str] = gist_desc - self.gist_id: Optional[str] = gist_id - self.response_status_str: str = "" - self.header: Dict[str, str] = { - "Authorization": f"token {secret_key}", - "Accept": "application/vnd.github+json", - } - self.end_point: str = "https://api.github.com/gists" - self.payload: Dict[str, Union[Dict[str, Dict[str, str]], bool, str]] = {} - self.logger: Logger = Logger(send_log=do_logging) - - def __send_http_request( - self, - end_point: Optional[str] = None, - header: Optional[Dict[str, str]] = None, - params: Optional[str] = None, - no_header: bool = False, - ) -> Union[int, List[Dict[str, Union[str, Dict[str, str], None, bool, int]]]]: - """sends the actual http request with params - - Sends the actual http request with metadata and returns the response. - - Args: - end_point: optional endpoint string. - header: optional header for the request. - params: Optional paramters for the request. - - Returns: - Returns http call return-code or json formatted response. - """ - if end_point is None: - end_point = self.end_point - if header is None: - header = self.header - http_operation = getattr(requests, self.http_operation) - response: requests.models.Response = ( - http_operation(url=end_point, data=params, headers=header) - if not no_header - else http_operation(url=end_point, data=params) - ) - self.response_status_str = self.__response_validator(response) - return_var: Union[ - int, List[Dict[str, Union[str, Dict[str, str], bool, int, None]]] - ] - if self.http_operation in ["get", "post"]: - response_hashtable = json.loads(response.content.decode("utf-8")) - if self.http_operation == "get": - return_var = response_hashtable - else: - return_var = response_hashtable.get("html_url") - elif self.http_operation in ["patch", "delete"]: - return_var = response.status_code - - return return_var - - def __response_validator(self, response: requests.models.Response) -> str: - """Validates the response - - Takes http response as argument and checks to see if its valid,returns the response string,otherwise raises - respective exception. - - Args: - response: Http response - Returns: - Response status string. - - """ - try: - http_response_codes: Misc.HttpResponseCodes = Misc.http_response_codes - response_str: str - exception_obj_dict: Dict[str, Misc.OptionalException] - response_str, exception_obj_dict = tuple( - http_response_codes[response.status_code].items() # type: ignore - )[0] - exception_obj = exception_obj_dict.get("exception_obj") - response_action: Callable[ - [Misc.OptionalException], None - ] = http_response_codes.get( # type: ignore - "exception_action" - ) - response_action(exception_obj) - except KeyError: - raise CegExceptions.InternalException( - "Undefined Response!.please open an issue on github." - ) - - return response_str - - def list( - self, - end_point: Optional[str] = None, - header: Optional[Dict[str, str]] = None, - no_header: bool = False, - ) -> Optional[List[Dict[str, str]]]: - """lists public/private gists for authenticated user. - - Performs GET operation on endpoint and retrieves all the public/private gists and propagates the formatted - response to standard stream. - - Args: - end_point: optional endpoint string. - header: optional header for the request. - - Returns: - (Optionally) returns a list containing all gists. - """ - if end_point is None: - end_point = self.end_point - if header is None: - header = self.header - - hashtable_response: Union[ - int, List[Dict[str, Union[str, Dict[str, str], None, bool, int]]] - ] = ( - self.__send_http_request(end_point, header) - if not no_header - else self.__send_http_request(end_point, no_header=True) - ) - - write_stdout: WriteStdout = WriteStdout(self.to_stdout) - for gist_no, gist_hashtable in enumerate(hashtable_response): # type: ignore - write_stdout( - f"Gist#{gist_no}", - is_rule=True, - silent=not self.to_stdout, - text_style="[cyan bold]", - ) - write_stdout(f"GistId: {gist_hashtable.get('id')}", text_style="[yellow]") - write_stdout( - f"Publicity: {'Public' if gist_hashtable.get('public') else 'Private'}", - text_style="[blue]", - ) - write_stdout( - f"Description: {gist_hashtable.get('description')}", - text_style="[grey58]", - ) - file_tree: Tree = Tree( - "[bold magenta]Files", guide_style="green underline2" - ) - for file_name, file_hashtable in gist_hashtable.get( - "files" - ).items(): # type: ignore - file_tree_branch: Tree = file_tree.add(file_name) - file_tree_branch.add( - f"Filesize: {file_hashtable.get('size')} bytes" # type: ignore - ) - file_tree_branch.add( - f"Language: {file_hashtable.get('language')}" # type: ignore - ) - file_tree_branch.add(f"Created at: {gist_hashtable.get('created_at')}") - file_tree_branch.add(f"Updated at: {gist_hashtable.get('updated_at')}") - write_stdout(file_tree, iteration_complete=True) - write_stdout(is_rule=True, silent=not self.to_stdout) - if not self.to_stdout: - return write_stdout.stream_cache - # welp mypy wants explicit return statement - else: - return None - - def get(self, **kwargs) -> None: - """Download gists using gist-ids as argument. - - This method downloads all the gists given on cli. backup() also uses this method internally. - - Args: - **kwargs = Auxiliary arbitrary keyword arguments. - """ - gist_id: Optional[str] - do_logging: Optional[bool] - bypass_recursion: Optional[bool] - gist_id = kwargs.get("gist_id") - do_logging = kwargs.get("logging_status") - bypass_recursion = kwargs.get("bypass_recursion") - hashtable_response: Union[ - int, List[Dict[str, Union[str, Dict[str, str], None, bool, int]]] - ] - if gist_id is not None: - self.logger.info( - f"Inquiring for gist with id '{gist_id}'", send_log=do_logging - ) - hashtable_response = self.__send_http_request(self.end_point, self.header) - gist_ids: List[str] = [ - gist.get("id") # type: ignore - for gist in hashtable_response # type: ignore - ] - - gist_id_list: List[str] = ( - gist_ids if self.is_recursive_op else self.arg_val # type: ignore - ) - if not bypass_recursion: - do_logging = False if self.is_recursive_op else True - for gist in gist_id_list: - self.get(gist_id=gist, bypass_recursion=True, logging_status=do_logging) - return None - try: - gist_id_index: int = gist_ids.index(gist_id) - except ValueError: - raise CegExceptions.ResourceNotFound("The Inquired Gist was not found!") - self.logger.info("Gist Found!", send_log=do_logging) - gist_files: List[Tuple[str, str]] = [ - (key, value) - for key, val in hashtable_response[gist_id_index] # type: ignore - .get("files") - .items() - if (value := val.get("raw_url")) is not None # type: ignore - ] - file_handler: FileHandler = FileHandler(dir_name=gist_id) - self.logger.info( - "Downloading and organizing all the files!", send_log=do_logging - ) - for single_gist in gist_files: - try: - file_name: str - file_url: str - file_name, file_url = single_gist - file_content = requests.get(url=file_url).content.decode("utf-8") - file_handler.write(file_name, file_content) - except requests.exceptions.ConnectionError: - file_handler.return_code = 1 - raise requests.exceptions.ConnectionError( - "Connection Error!,please check your internet connection." - ) - self.logger.info("Sucessfully downloaded the gist!", send_log=do_logging) - - def post(self, **kwargs) -> Optional[str]: - """Create gists. - - Creates arbitrary number of gists depending on files given on cli. - - Args: - **kwargs: Auxiliary arbitrary keyword arguments. - - Returns: - (Optionally) return html url of the newly created gist - """ - is_patch: Optional[bool] - new_filenames: Optional[Dict[str, str]] - - is_patch = kwargs.get("is_patch") - new_filenames = kwargs.get("new_filenames") - - op_success_msg: Dict[str, str] = {"post": "published", "patch": "updated"} - all_files_validate: bool = True - files_to_ignore: List[str] = [] - file_to_content_map: Dict[str, Dict[str, str]] = {} - for file in self.arg_val: # type: ignore - if not os.path.exists(file): - self.logger.info( - f"{file} not found in given path,opening in default editor.." - ) - ret_code: int = open_file(file) - if ret_code != 0: - self.logger.warning( - f"An Error occured while opening '{file}' in default editor." - ) - self.logger.info(f"Validating filename for '{file}'") - file_basename: str = os.path.basename(file) - validated: bool = gist_filename_validated(file_basename) - if not validated: - all_files_validate = False - self.logger.warning(f"{file} will be ignored!") - files_to_ignore.append(file) - continue - with open(file, "r") as r_obj: - file_content: str = r_obj.read() - file_to_content_map.update({file_basename: {"content": file_content}}) - if is_patch and new_filenames.get(file_basename): # type: ignore - file_to_content_map[file_basename].update( - {"filename": new_filenames.get(file_basename)} # type: ignore - ) - - if not all_files_validate: - self.logger.warning( - "One or more files were found to have filenames that are prohibited by github and hence will be ignored." - ) - self.payload.update({"files": file_to_content_map}) - if is_patch is None: - self.payload.update({"public": not self.gist_no_public}) - if self.gist_description: - self.payload.update({"description": self.gist_description}) - gist_html_url = self.__send_http_request(params=json.dumps(self.payload)) - self.logger.info(f"Sucessfully {op_success_msg[self.http_operation]} the gist!") - if self.http_operation == "post": - if self.to_stdout: - self.logger.info(f"Gist Url: {gist_html_url}") - else: - return gist_html_url # type: ignore - return None - - def patch(self) -> None: - """Modify an existing gist.""" - if self.gist_id is None: - raise CegExceptions.InsufficientSubArguments("--gist-id missing!") - new_filename_map: Dict[str, str] = {} - self.end_point += "/" + self.gist_id - - for file_index, file in enumerate(self.arg_val): # type: ignore - try: - oldname, newname = file.split("->") - oldname_base: str = os.path.basename(oldname) - new_filename_map.update({oldname_base: newname}) - self.arg_val[file_index] = oldname # type: ignore - except ValueError: - pass - self.post(is_patch=True, new_filenames=new_filename_map) - - def delete(self) -> None: - """Delete an existing gist.""" - endpoint_copy: str = self.end_point - for gist in self.arg_val: # type: ignore - self.logger.info(f"Searching and deleting gist with id '{gist[:4]}...'") - self.end_point = "{}/{}".format(endpoint_copy, gist) # type: ignore - self.__send_http_request() - self.logger.info("Gist deleted sucessfully!.") - - def backup(self) -> None: - """Create a local backup of all the gists.""" - self.logger.info("Backing up all gists to local media...") - dir_handler: FileHandler = FileHandler("GIST-BACKUP") - os.chdir("GIST-BACKUP") - try: - self.get() - except requests.exceptions.ConnectionError: - dir_handler.return_code = 1 - raise requests.exceptions.ConnectionError( - "Connection Error!,please check your internet connection." - ) - else: - self.logger.info("Backup successfull!") - - def list_other(self) -> Optional[List[Dict[str, str]]]: - """Get gists for other users - - This method simply mutates the endpoint and nulls the auth headers and simply calls the list() method. - Args: - user_name: github username for the other user. - Returns: - (Optionally) returns a list containing all gists. - """ - user_gist_info: Optional[List[Dict[str, str]]] = self.list( - end_point=f"https://api.github.com/users/{self.arg_val}/gists", - no_header=True, - ) - if not self.to_stdout: - return user_gist_info - else: - return None - - def perform_operation(self) -> int: - """Wrapper for all the methods - - Performs all the http requests,sorts out data and - does all the pretty printing using rich. - - Returns: - Return code used on exit,indicating success or failure. - - Raises: - ConnectionError: raised due to connection error while sending the http request. - BadCredentials: raised due to bad github secret key. - ResourceNotFound: raised due to inavailability of inquired resource. - """ - try: - if self.arg_val is None and not self.is_recursive_op: - self.list() - elif self.is_other: - self.list_other() - elif self.is_recursive_op: - self.backup() - else: - http_intrinsics = getattr(self, self.http_operation) - http_intrinsics() - except ( - CegExceptions.BadCredentials, - CegExceptions.ResourceNotFound, - CegExceptions.UnprocessableRequest, - CegExceptions.InsufficientSubArguments, - requests.exceptions.ConnectionError, - ): - self.logger.exception("An Error has occured!") - return GenericReturnCodes.FAILURE - except Exception: - self.logger.exception( - "An internal exception has risen!.please open an issue if you think this is a bug" - ) - return GenericReturnCodes.FAILURE - else: - return GenericReturnCodes.SUCCESS +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" Main implementation of ceg """ + +import re +import os +import sys +import json +import requests +from .logger import Logger +from rich.tree import Tree +from rich.console import Console +from .exceptions import GenericReturnCodes, CegExceptions +from .misc import Misc, FileHandler, open_file, gist_filename_validated +from typing import List, Tuple, Dict, Optional, Union, Callable, Any + +console: Console = Console() + +__all__ = ("Ceg",) + + +class AuxSequence(List): + """An Auxiliary sequence. + + An Auxiliary Sequence that structurally saves stdout for + delivering api. + + Attributes: + single_item_dict: A dictionary containing elements of single gist. + """ + + def __init__(self) -> None: + self.single_item_dict: Dict[str, Union[str, List[str]]] = {} + + def append( + self, + writable: Optional[Union[str, Tree, Dict[str, List[str]]]], + iteration_complete: bool = False, + ) -> None: + """Overloads the append method of List + + Receives (unsanitized) data which is added to + a mapping,which is then appended to the + sequence itself. + + Args: + writable: Actual stream which is to be stored. + iteration_complete: A boolean acting as a delimiter for gists. + """ + if isinstance(writable, Dict): + self.single_item_dict.update(writable) + else: + key, val = writable.split(":") # type: ignore + self.single_item_dict.update({key: val}) + if iteration_complete: + single_item_dict_copy = self.single_item_dict.copy() + super().append(single_item_dict_copy) + self.single_item_dict.clear() + + +class WriteStdout: + """Write stdout to an initialized resource. + + Implements the "write stdout" functionality.It writes the stdout to an initialized resource. + + Attributes: + to_stdout: A bool indicating whether to write to stdout or not. + stream_cache: A list containing all gists info. + write_stdout: A Callable which either writes to stdout or stream_cache. + """ + + def __init__(self, to_stdout: bool) -> None: + """Inits WriteStdout with appropriate attributes""" + self.to_stdout: bool = to_stdout + self.stream_cache: AuxSequence = AuxSequence() + self.write_stdout: Callable[ + [Optional[Union[str, Tree, Dict[str, List[str]]]], bool], None + ] + # mypy was being bitchy when using ternary operator so ... + # related issue: https://github.com/python/mypy/issues/10740 + if self.to_stdout: + self.write_stdout = lambda writable, _: console.print(writable) + else: + self.write_stdout = self.stream_cache.append + + def __call__( + self, + writable: Optional[Union[str, Tree, Dict[str, List[str]]]] = None, + is_rule: bool = False, + silent: bool = False, + text_style: Optional[str] = None, + iteration_complete: bool = False, + ) -> None: + """Implements dunder call + + Calls the actual write_stdout method after evaluation. + + Args: + writable: A string or Tree object. + is_rule: A boolean indicating if the writable is a rule object. + is_tree: A boolean indicating if the writable is a Tree object. + silent: A boolean indicating whether to waste the write_stdout + call(mostly while self.to_stdout is False and we don't + want to print the rule). + text_style: Contains the styling string. + + """ + if isinstance(writable, Tree) and not self.to_stdout: + tree_dict: Dict[str, List[str]] = {"Files": []} + retreive_file: Callable[ + [Tree], str + ] = lambda tree_obj: tree_obj.label # type: ignore + tree_files: List[str] = list(map(retreive_file, writable.children)) + tree_dict["Files"] = tree_files + writable = tree_dict + if not silent: + writable = text_style + writable if text_style and self.to_stdout else writable # type: ignore + self.write_stdout( + writable, iteration_complete + ) if not is_rule else console.rule( + writable # type: ignore + ) + + +class Ceg: + """Main implementation of ceg. + + The Ceg class actually contains the main implementation + of the utility. + + Attributes: + http_operation: A Callable that actually performs the http calls. + arg_val: A string that is used as argument value as well as a switch for operation resolution. + is_recursive_op: A boolean indicating if the operation is recursive. + header: A Dict containing the HTTP header. + end_point: Endpoint for api calls. + payload: payload containing data for post requests. + to_stdout: boolean indicating whether to send data to stdout. + gist_no_public: boolean indicating if a gist is private. + is_other: boolean indicating if to list gists for unauth user. + gist_description: String containing gist description which can be used in patch(),post(). + gist_id: String containing gist-id. + response_status_str: Contains HTTP call response status in string format. + logger: Logger object. + ceg_get_namespace: a namespace containing states and responses relating to Ceg.get() + """ + + def __init__( + self, + operation: str, + arg_value: Optional[Union[str, Tuple[str, ...]]], + is_recursive_operation: bool, + is_other_user: bool, + secret_key: Optional[str], + do_logging: bool, + gist_no_public: bool, + gist_desc: Optional[str], + gist_id: Optional[str], + ) -> None: + """Inits Ceg with appropriate attributes""" + self.http_operation: str = operation + self.arg_val: Optional[Union[str, Tuple[str, ...]]] = arg_value + self.to_stdout: bool = do_logging + self.is_recursive_op: bool = is_recursive_operation + self.is_other: bool = is_other_user + self.gist_no_public: bool = gist_no_public + self.gist_description: Optional[str] = gist_desc + self.gist_id: Optional[str] = gist_id + self.response_status_str: str = "" + self.header: Dict[str, str] = { + "Authorization": f"token {secret_key}", + "Accept": "application/vnd.github+json", + } + self.end_point: str = "https://api.github.com/gists" + self.payload: Dict[str, Union[Dict[str, Dict[str, str]], bool, str]] = {} + self.logger: Logger = Logger(send_log=do_logging) + # ceg_get_namespace["response"] has a ridiculous annotation + # so better off using `Any` to please mypy + self.ceg_get_namespace: Dict[str, Union[bool, Any]] = { + "has_response": False, + "response": None, + } + + def __send_http_request( + self, + end_point: Optional[str] = None, + header: Optional[Dict[str, str]] = None, + params: Optional[str] = None, + no_header: bool = False, + ) -> Union[int, List[Dict[str, Union[str, Dict[str, str], None, bool, int]]]]: + """sends the actual http request with params + + Sends the actual http request with metadata and returns the response. + + Args: + end_point: optional endpoint string. + header: optional header for the request. + params: Optional paramters for the request. + + Returns: + Returns http call return-code or json formatted response. + """ + if end_point is None: + end_point = self.end_point + if header is None: + header = self.header + + http_operation = getattr(requests, self.http_operation) + response: requests.models.Response = ( + http_operation(url=end_point, data=params, headers=header) + if not no_header + else http_operation(url=end_point, data=params) + ) + self.response_status_str = self.__response_validator(response) + return_var: Union[ + int, List[Dict[str, Union[str, Dict[str, str], bool, int, None]]] + ] + if self.http_operation in ["get", "post"]: + response_hashtable = json.loads(response.content.decode("utf-8")) + if self.http_operation == "get": + return_var = response_hashtable + else: + return_var = response_hashtable.get("html_url") + elif self.http_operation in ["patch", "delete"]: + return_var = response.status_code + + return return_var + + def __response_validator(self, response: requests.models.Response) -> str: + """Validates the response + + Takes http response as argument and checks to see if its valid,returns the response string,otherwise raises + respective exception. + + Args: + response: Http response + Returns: + Response status string. + + """ + try: + http_response_codes: Misc.HttpResponseCodes = Misc.http_response_codes + response_str: str + exception_obj_dict: Dict[str, Misc.OptionalException] + response_str, exception_obj_dict = tuple( + http_response_codes[response.status_code].items() # type: ignore + )[0] + exception_obj = exception_obj_dict.get("exception_obj") + response_action: Callable[ + [Misc.OptionalException], None + ] = http_response_codes.get( # type: ignore + "exception_action" + ) + response_action(exception_obj) + except KeyError: + raise CegExceptions.InternalException( + "Undefined Response!.please open an issue on github." + ) + + return response_str + + def list( + self, + end_point: Optional[str] = None, + header: Optional[Dict[str, str]] = None, + no_header: bool = False, + ) -> Optional[List[Dict[str, str]]]: + """lists public/private gists for authenticated user. + + Performs GET operation on endpoint and retrieves all the public/private gists and propagates the formatted + response to standard stream. + + Args: + end_point: optional endpoint string. + header: optional header for the request. + + Returns: + (Optionally) returns a list containing all gists. + """ + if end_point is None: + end_point = self.end_point + if header is None: + header = self.header + + hashtable_response: Union[ + int, List[Dict[str, Union[str, Dict[str, str], None, bool, int]]] + ] = ( + self.__send_http_request(end_point, header) + if not no_header + else self.__send_http_request(end_point, no_header=True) + ) + + write_stdout: WriteStdout = WriteStdout(self.to_stdout) + for gist_no, gist_hashtable in enumerate(hashtable_response): # type: ignore + write_stdout( + f"Gist#{gist_no}", + is_rule=True, + silent=not self.to_stdout, + text_style="[cyan bold]", + ) + write_stdout(f"GistId: {gist_hashtable.get('id')}", text_style="[yellow]") + write_stdout( + f"Publicity: {'Public' if gist_hashtable.get('public') else 'Private'}", + text_style="[blue]", + ) + write_stdout( + f"Description: {gist_hashtable.get('description')}", + text_style="[grey58]", + ) + file_tree: Tree = Tree( + "[bold magenta]Files", guide_style="green underline2" + ) + for file_name, file_hashtable in gist_hashtable.get( + "files" + ).items(): # type: ignore + file_tree_branch: Tree = file_tree.add(file_name) + file_tree_branch.add( + f"Filesize: {file_hashtable.get('size')} bytes" # type: ignore + ) + file_tree_branch.add( + f"Language: {file_hashtable.get('language')}" # type: ignore + ) + file_tree_branch.add(f"Created at: {gist_hashtable.get('created_at')}") + file_tree_branch.add(f"Updated at: {gist_hashtable.get('updated_at')}") + write_stdout(file_tree, iteration_complete=True) + write_stdout(is_rule=True, silent=not self.to_stdout) + if not self.to_stdout: + return write_stdout.stream_cache + # welp mypy wants explicit return statement + else: + return None + + def get(self, **kwargs) -> None: + """Download gists using gist-ids as argument. + + This method downloads all the gists given on cli. backup() also uses this method internally. + + Args: + **kwargs = Auxiliary arbitrary keyword arguments. + """ + gist_id: Optional[str] + do_logging: Optional[bool] + bypass_recursion: Optional[bool] + no_header: bool = False + gist_id = kwargs.get("gist_id") + do_logging = kwargs.get("logging_status") + bypass_recursion = kwargs.get("bypass_recursion") + hashtable_response: Union[ + int, List[Dict[str, Union[str, Dict[str, str], None, bool, int]]] + ] + match_str: Union[List, str] = ( + self.arg_val if isinstance(self.arg_val, str) else self.arg_val[0] # type: ignore + ) + if username_match := re.match("user:\w+", match_str): # type: ignore + if not re.match( + "https:\/\/api\.github\.com\/users\/\w+\/gists", self.end_point + ): + self.end_point = re.sub( + "gists", + f"users/{username_match.group().split(':')[1]}/gists", + self.end_point, + ) + no_header = True + if isinstance(self.arg_val, List): + self.arg_val.pop(0) + + if gist_id is not None: + self.logger.info( + f"Inquiring for gist with id '{gist_id}'", send_log=do_logging + ) + if not self.ceg_get_namespace["has_response"]: + self.ceg_get_namespace["response"] = self.__send_http_request( + self.end_point, self.header, no_header=no_header + ) + self.ceg_get_namespace["has_response"] = True + + hashtable_response = self.ceg_get_namespace["response"] + + gist_ids: List[str] = [ + gist.get("id") # type: ignore + for gist in hashtable_response # type: ignore + ] + + gist_id_list: List[str] = ( + gist_ids if self.is_recursive_op else self.arg_val # type: ignore + ) + if not bypass_recursion: + do_logging = False if self.is_recursive_op else True + for gist in gist_id_list: + self.get(gist_id=gist, bypass_recursion=True, logging_status=do_logging) + return None + try: + gist_id_index: int = gist_ids.index(gist_id) + except ValueError: + raise CegExceptions.ResourceNotFound("The Inquired Gist was not found!") + self.logger.info("Gist Found!", send_log=do_logging) + gist_files: List[Tuple[str, str]] = [ + (key, value) + for key, val in hashtable_response[gist_id_index] # type: ignore + .get("files") + .items() + if (value := val.get("raw_url")) is not None # type: ignore + ] + file_handler: FileHandler = FileHandler(dir_name=gist_id) + self.logger.info( + "Downloading and organizing all the files!", send_log=do_logging + ) + for single_gist in gist_files: + try: + file_name: str + file_url: str + file_name, file_url = single_gist + file_content = requests.get(url=file_url).content.decode("utf-8") + file_handler.write(file_name, file_content) + except requests.exceptions.ConnectionError: + file_handler.return_code = 1 + raise requests.exceptions.ConnectionError( + "Connection Error!,please check your internet connection." + ) + self.logger.info("Sucessfully downloaded the gist!", send_log=do_logging) + + def post(self, **kwargs) -> Optional[str]: + """Create gists. + + Creates arbitrary number of gists depending on files given on cli. + + Args: + **kwargs: Auxiliary arbitrary keyword arguments. + + Returns: + (Optionally) return html url of the newly created gist + """ + is_patch: Optional[bool] + new_filenames: Optional[Dict[str, str]] + + is_patch = kwargs.get("is_patch") + new_filenames = kwargs.get("new_filenames") + + op_success_msg: Dict[str, str] = {"post": "published", "patch": "updated"} + all_files_validate: bool = True + files_to_ignore: List[str] = [] + file_to_content_map: Dict[str, Dict[str, str]] = {} + for file in self.arg_val: # type: ignore + if not os.path.exists(file): + self.logger.info( + f"{file} not found in given path,opening in default editor.." + ) + ret_code: int = open_file(file) + if ret_code != 0: + self.logger.warning( + f"An Error occured while opening '{file}' in default editor." + ) + files_to_ignore.append(file) + continue + self.logger.info(f"Validating filename for '{file}'") + file_basename: str = os.path.basename(file) + validated: bool = gist_filename_validated(file_basename) + if not validated: + all_files_validate = False + self.logger.warning(f"{file} will be ignored!") + files_to_ignore.append(file) + continue + with open(file, "r") as r_obj: + file_content: str = r_obj.read() + file_to_content_map.update({file_basename: {"content": file_content}}) + if is_patch and new_filenames.get(file_basename): # type: ignore + file_to_content_map[file_basename].update( + {"filename": new_filenames.get(file_basename)} # type: ignore + ) + + if not all_files_validate: + self.logger.warning( + "One or more files were found to have filenames that are prohibited by github and hence will be ignored." + ) + self.payload.update({"files": file_to_content_map}) + if is_patch is None: + self.payload.update({"public": not self.gist_no_public}) + if self.gist_description: + self.payload.update({"description": self.gist_description}) + gist_html_url = self.__send_http_request(params=json.dumps(self.payload)) + self.logger.info(f"Sucessfully {op_success_msg[self.http_operation]} the gist!") + if self.http_operation == "post": + if self.to_stdout: + self.logger.info(f"Gist Url: {gist_html_url}") + else: + return gist_html_url # type: ignore + return None + + def patch(self) -> None: + """Modify an existing gist.""" + if self.gist_id is None: + raise CegExceptions.InsufficientSubArguments("--gist-id missing!") + new_filename_map: Dict[str, str] = {} + self.end_point += "/" + self.gist_id + + for file_index, file in enumerate(self.arg_val): # type: ignore + try: + oldname, newname = file.split("->") + oldname_base: str = os.path.basename(oldname) + new_filename_map.update({oldname_base: newname}) + self.arg_val[file_index] = oldname # type: ignore + except ValueError: + pass + self.post(is_patch=True, new_filenames=new_filename_map) + + def delete(self) -> None: + """Delete an existing gist.""" + endpoint_copy: str = self.end_point + for gist in self.arg_val: # type: ignore + self.logger.info(f"Searching and deleting gist with id '{gist[:4]}...'") + self.end_point = "{}/{}".format(endpoint_copy, gist) # type: ignore + self.__send_http_request() + self.logger.info("Gist deleted sucessfully!.") + + def backup(self) -> None: + """Create a local backup of all the gists.""" + self.logger.info("Backing up all gists to local media...") + dir_handler: FileHandler = FileHandler("GIST-BACKUP") + os.chdir("GIST-BACKUP") + try: + self.get() + except requests.exceptions.ConnectionError: + raise requests.exceptions.ConnectionError( + "Connection Error!,please check your internet connection." + ) + except Exception: + os.chdir("../") + dir_handler.return_code = 1 + raise sys.exc_info()[1] # type: ignore + else: + self.logger.info("Backup successfull!") + + def list_other(self) -> Optional[List[Dict[str, str]]]: + """Get gists for other users + + This method simply mutates the endpoint and nulls the auth headers and simply calls the list() method. + Args: + user_name: github username for the other user. + Returns: + (Optionally) returns a list containing all gists. + """ + prefix: str + suffix: str + try: + prefix, suffix = self.arg_val.split(":") # type: ignore + except ValueError: + raise AssertionError( + f"Expected username argument of pattern `user:user_name` but got `{self.arg_val}`" + ) + assert prefix == "user" + user_gist_info: Optional[List[Dict[str, str]]] = self.list( + end_point=f"https://api.github.com/users/{suffix}/gists", + no_header=True, + ) + if not self.to_stdout: + return user_gist_info + else: + return None + + def perform_operation(self) -> int: + """Wrapper for all the methods + + Performs all the http requests,sorts out data and + does all the pretty printing using rich. + + Returns: + Return code used on exit,indicating success or failure. + + Raises: + ConnectionError: raised due to connection error while sending the http request. + BadCredentials: raised due to bad github secret key. + ResourceNotFound: raised due to inavailability of inquired resource. + """ + try: + if self.arg_val == "self" and not self.is_recursive_op: + self.list() + elif self.is_other: + self.list_other() + elif self.is_recursive_op: + self.backup() + else: + http_intrinsics = getattr(self, self.http_operation) + http_intrinsics() + except ( + CegExceptions.BadCredentials, + CegExceptions.ResourceNotFound, + CegExceptions.UnprocessableRequest, + CegExceptions.InsufficientSubArguments, + requests.exceptions.ConnectionError, + ): + self.logger.exception("An Error has occured!") + return GenericReturnCodes.FAILURE + except Exception: + self.logger.exception( + "An internal exception has risen!.please open an issue if you think this is a bug" + ) + return GenericReturnCodes.FAILURE + else: + return GenericReturnCodes.SUCCESS diff --git a/ceg/cli.py b/ceg/cli.py old mode 100755 new mode 100644 index 2ba3114..48c1951 --- a/ceg/cli.py +++ b/ceg/cli.py @@ -1,81 +1,83 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" Main cli """ - -import sys -from .ceg import Ceg -from .misc import Misc -from .arg_parser import ArgumentParser -from typing import Optional, Any - - -def ceg_cli() -> None: - """First subroutine to run on execution of utility. - - The main function which reccords the arguments - and also partially handles them. - """ - arg_parse_obj: ArgumentParser = ArgumentParser() - parsed_args = arg_parse_obj.reccord_arguments() - http_operation: str - gist_description: Optional[str] = None - gist_id: Optional[str] = None - is_recursive: bool = False - is_other: bool = False - logging_status: bool = True - gist_no_public: bool = False - argument_value: Optional[Any] - - for arg, arg_val in vars(parsed_args).items(): - if arg == "secret_key" and arg_val: - Misc.secret_key = arg_val - elif arg == "no_logging" and arg_val: - logging_status = False - elif arg == "no_public" and arg_val: - gist_no_public = True - elif arg == "description" and arg_val: - gist_description = arg_val - elif arg == "gist_id" and arg_val: - gist_id = arg_val - elif arg in Misc.http_intrinsics and arg_val: - http_operation = arg - argument_value = arg_val - elif arg in ["list", "backup", "list_other"] and arg_val: - http_operation = "get" - is_recursive = True if arg == "backup" else False - is_other = True if arg == "list_other" else False - argument_value = None if arg in ["list", "backup"] else arg_val - - ceg_obj = Ceg( - operation=http_operation, - arg_value=argument_value, - is_recursive_operation=is_recursive, - is_other_user=is_other, - secret_key=Misc.secret_key, - do_logging=logging_status, - gist_no_public=gist_no_public, - gist_desc=gist_description, - gist_id=gist_id, - ) - return_code: int = ceg_obj.perform_operation() - sys.exit(return_code) +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" Main cli """ + +import sys +from .ceg import Ceg +from .misc import Misc +from .arg_parser import ArgumentParser +from typing import Optional, Any + + +def ceg_cli() -> None: + """First subroutine to run on execution of utility. + + The main function which reccords the arguments + and also partially handles them. + """ + arg_parse_obj: ArgumentParser = ArgumentParser() + parsed_args = arg_parse_obj.reccord_arguments() + http_operation: str + gist_description: Optional[str] = None + gist_id: Optional[str] = None + is_recursive: bool = False + is_other: bool = False + logging_status: bool = True + gist_no_public: bool = False + argument_value: Optional[Any] + + for arg, arg_val in vars(parsed_args).items(): + if arg == "secret_key" and arg_val: + Misc.secret_key = arg_val + elif arg == "no_logging" and arg_val: + logging_status = False + elif arg == "no_public" and arg_val: + gist_no_public = True + elif arg == "description" and arg_val: + gist_description = arg_val + elif arg == "gist_id" and arg_val: + gist_id = arg_val + elif arg in Misc.http_intrinsics and arg_val: + http_operation = arg + argument_value = arg_val + elif arg in ("list", "backup") and arg_val: + http_operation = "get" + is_recursive = True if arg == "backup" else False + is_other = True if (arg == "list" and arg_val != "self") else False + argument_value = ( + None if (arg in ("list", "backup") and arg_val is None) else arg_val + ) + + ceg_obj = Ceg( + operation=http_operation, + arg_value=argument_value, + is_recursive_operation=is_recursive, + is_other_user=is_other, + secret_key=Misc.secret_key, + do_logging=logging_status, + gist_no_public=gist_no_public, + gist_desc=gist_description, + gist_id=gist_id, + ) + return_code: int = ceg_obj.perform_operation() + sys.exit(return_code) diff --git a/ceg/exceptions.py b/ceg/exceptions.py old mode 100755 new mode 100644 index a609894..9f09fcc --- a/ceg/exceptions.py +++ b/ceg/exceptions.py @@ -1,82 +1,82 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" Cegs exceptions """ - - -class CegExceptions: - """All of Cegs exceptions.""" - - class BadCredentials(Exception): - """raised when a bad response like bad auth is risen.""" - - def __init__(self, msg: str = "Bad Credentials!") -> None: - super().__init__(msg) - - class ResourceNotFound(Exception): - """raised when inquired resource is not found.""" - - def __init__(self, msg: str = "Inquired Resource not found!") -> None: - super().__init__(msg) - - class ForbiddenResource(Exception): - """raised when forbidden resource is inquired.""" - - def __init__(self, msg: str = "Forbidden Resource!") -> None: - super().__init__(msg) - - class UnprocessableRequest(Exception): - """raised when a syntatically correct (tho possibly semantically wrong) request is unprocessable by the server.""" - - def __init__(self, msg: str = "Unprocessable Request!") -> None: - super().__init__(msg) - - class InsufficientSubArguments(Exception): - """raised when sub-arguments are missing from argument list. - - Due to nature of handling of arguments from cli,missing sub-arguments - are not catched by argparse. - """ - - def __init__(self, msg: str) -> None: - super().__init__(msg) - - class InternalException(Exception): - """raised due to internal errors.""" - - def __init__(self, msg: str = "Unprocessable Resource") -> None: - super().__init__(msg) - - -class GenericReturnCodes: - """Generic Return Codes - - Contains the commandline return codes. - - Attributes: - SUCCESS: return code for successfull operation. - FAILURE: return code incase an error occurs. - """ - - # TODO: for [v0.2.0]: return different exit codes for better script compatibility - SUCCESS: int = 0 - FAILURE: int = 1 +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" Cegs exceptions """ + + +class CegExceptions: + """All of Cegs exceptions.""" + + class BadCredentials(Exception): + """raised when a bad response like bad auth is risen.""" + + def __init__(self, msg: str = "Bad Credentials!") -> None: + super().__init__(msg) + + class ResourceNotFound(Exception): + """raised when inquired resource is not found.""" + + def __init__(self, msg: str = "Inquired Resource not found!") -> None: + super().__init__(msg) + + class ForbiddenResource(Exception): + """raised when forbidden resource is inquired.""" + + def __init__(self, msg: str = "Forbidden Resource!") -> None: + super().__init__(msg) + + class UnprocessableRequest(Exception): + """raised when a syntatically correct (tho possibly semantically wrong) request is unprocessable by the server.""" + + def __init__(self, msg: str = "Unprocessable Request!") -> None: + super().__init__(msg) + + class InsufficientSubArguments(Exception): + """raised when sub-arguments are missing from argument list. + + Due to nature of handling of arguments from cli,missing sub-arguments + are not catched by argparse. + """ + + def __init__(self, msg: str) -> None: + super().__init__(msg) + + class InternalException(Exception): + """raised due to internal errors.""" + + def __init__(self, msg: str = "Unprocessable Resource") -> None: + super().__init__(msg) + + +class GenericReturnCodes: + """Generic Return Codes + + Contains the commandline return codes. + + Attributes: + SUCCESS: return code for successfull operation. + FAILURE: return code incase an error occurs. + """ + + # TODO: for [v0.2.0]: return different exit codes for better script compatibility + SUCCESS: int = 0 + FAILURE: int = 1 diff --git a/ceg/logger.py b/ceg/logger.py old mode 100755 new mode 100644 index 3e82703..24c061c --- a/ceg/logger.py +++ b/ceg/logger.py @@ -1,76 +1,76 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" Enhanced Logger for Ceg """ - -import logging -from rich.logging import RichHandler -from typing import Union, Any, no_type_check - - -class Logger(logging.getLoggerClass()): - """Logger object responsible for logging all the events. - - Inherits from Logger class of stdlib logging library - and implements the log propagation validation,i.e whether - logs should be logged to the stdout/stderr. - - Attributes: - send_log = boolean indicating whether to log or not - loglevel = indicates the effective log level. - dateformat = string represrnting datetime format. - handler = Handler used by the logger. - """ - - def __init__( - self, - send_log: bool = True, - loglevel: Union[int, str] = logging.INFO, - dateformat: str = "[%X]", - ) -> None: - """Inits the Logger""" - self.formatter: logging.Formatter = logging.Formatter(datefmt=dateformat) - self.handler: RichHandler = RichHandler() - - self.handler.setFormatter(self.formatter) - super().__init__(name=__name__, level=logging.INFO) - super().addHandler(self.handler) - self.setLevel(logging._checkLevel(loglevel)) # type: ignore - if not send_log: - logging.disable(logging.CRITICAL) - - @no_type_check - def info(self, msg: str, send_log: bool = True, *args: Any, **kwargs: Any) -> None: - if send_log: - super().info(msg, *args, **kwargs) - - @no_type_check - def warning( - self, msg: str, send_log: bool = True, *args: Any, **kwargs: Any - ) -> None: - if send_log: - super().warning(msg, *args, **kwargs) - - @no_type_check - def error(self, msg: str, send_log: bool = True, *args: Any, **kwargs: Any) -> None: - if send_log: - super().error(msg, *args, **kwargs) +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" Enhanced Logger for Ceg """ + +import logging +from rich.logging import RichHandler +from typing import Union, Any, no_type_check + + +class Logger(logging.getLoggerClass()): # type: ignore + """Logger object responsible for logging all the events. + + Inherits from Logger class of stdlib logging library + and implements the log propagation validation,i.e whether + logs should be logged to the stdout/stderr. + + Attributes: + send_log = boolean indicating whether to log or not + loglevel = indicates the effective log level. + dateformat = string represrnting datetime format. + handler = Handler used by the logger. + """ + + def __init__( + self, + send_log: bool = True, + loglevel: Union[int, str] = logging.INFO, + dateformat: str = "[%X]", + ) -> None: + """Inits the Logger""" + self.formatter: logging.Formatter = logging.Formatter(datefmt=dateformat) + self.handler: RichHandler = RichHandler() + + self.handler.setFormatter(self.formatter) + super().__init__(name=__name__, level=logging.INFO) + super().addHandler(self.handler) + self.setLevel(logging._checkLevel(loglevel)) # type: ignore + if not send_log: + logging.disable(logging.CRITICAL) + + @no_type_check + def info(self, msg: str, send_log: bool = True, *args: Any, **kwargs: Any) -> None: + if send_log: + super().info(msg, *args, **kwargs) + + @no_type_check + def warning( + self, msg: str, send_log: bool = True, *args: Any, **kwargs: Any + ) -> None: + if send_log: + super().warning(msg, *args, **kwargs) + + @no_type_check + def error(self, msg: str, send_log: bool = True, *args: Any, **kwargs: Any) -> None: + if send_log: + super().error(msg, *args, **kwargs) diff --git a/ceg/misc.py b/ceg/misc.py old mode 100755 new mode 100644 index 0aea924..f3969ed --- a/ceg/misc.py +++ b/ceg/misc.py @@ -1,209 +1,209 @@ -# ╔═══╗ -# ║╔═╗║ -# ║║ ╚╝╔══╗╔══╗ -# ║║ ╔╗║╔╗║║╔╗║ -# ║╚═╝║║║═╣║╚╝║ -# ╚═══╝╚══╝╚═╗║ -# ╔═╝║ -# ╚══╝ -# ©Justaus3r 2022 -# This file is part of "Ceg",a gist crud utility. -# Distributed under GPLV3 -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -""" Misc stuff dat i couldn't figure out where to put """ - -import os -import platform -import subprocess -from .exceptions import CegExceptions -from typing import List, Dict, Callable, Union, Optional, Type, TypeAlias - -__all__ = ("UtilInfo", "Misc") - - -class UtilInfo: - """Metainfo about utility. - - Class containing all the meta info about - the utility. - - Attributes: - UTIL_NAME: The utility name. - DESCRIPTION: Short description of the utility. - VERSION: Semantic verson of the utility. - """ - - UTIL_NAME: str = "ceg" - UTIL_USAGE: str = f"{UTIL_NAME} [options] [sub-arguments]" - EPILOG: str = """ -sub-arguments: - --post/-po - --no-public/-np switch gist visibility to private - - --description/-desc description for the gist - - --patch/-pa - --gist-id/-gi gist-id for the gist - -For more usage help, check out https://www.github.com/justaus3r/ceg/#examples""" - DESCRIPTION: str = "An all in one github's gist manager." - # Caution(message to myself): Be careful when updating the version because - # wrong updates can be a mess. - VERSION: str = "0.4.1" - - -def exception_executioner(exception_obj) -> None: - """Raises exception taken as am argument. - - Args: - exception_obj: THe Exception object. - - """ - if exception_obj: - raise exception_obj - - -class Misc: - """Misc stuff. - - Contains all the miscellaneous vars. - - Attributes: - http_intrinsics: List containing names of all http methods. - secret_key: Gitub Secret key extracted from env variable. - """ - - http_intrinsics: List[str] = ["get", "post", "patch", "delete"] - OptionalException: TypeAlias = Optional[ - Union[ - Type[CegExceptions.BadCredentials], - Type[CegExceptions.ForbiddenResource], - Type[CegExceptions.ResourceNotFound], - Type[CegExceptions.UnprocessableRequest], - ] - ] - HttpResponseCodes: TypeAlias = Dict[ - Union[int, str], - Union[ - Dict[str, Dict[str, OptionalException]], Callable[[OptionalException], None] - ], - ] - http_response_codes: HttpResponseCodes = { - 200: {"OK!": {"exception_obj": None}}, - 201: {"Gist Created Sucessfully!": {"exception_obj": None}}, - 204: {"OK!.No Response Recieved.": {"exception_obj": None}}, - 401: {"Bad Credentials!": {"exception_obj": CegExceptions.BadCredentials}}, - 403: { - "Forbidden Resource!": {"exception_obj": CegExceptions.ForbiddenResource} - }, - 404: {"Resource not found!": {"exception_obj": CegExceptions.ResourceNotFound}}, - 422: { - "Request Unprocessable!": { - "exception_obj": CegExceptions.UnprocessableRequest - } - }, - "exception_action": exception_executioner, - } - secret_key: Optional[str] = os.getenv("GITHUB_SECRET_KEY") - - -class FileHandler: - """File & Directory Handler. - - Responsible for creating gist directories, - files,writing content to them and removing - gist directories if found empty due to some - error. - - Attributes: - return_code: return code which determines whether to keep a directory or not. - """ - - def __init__(self, dir_name: str) -> None: - "Inits FileHandler with some directory name" - self.__dir_name: str = dir_name - self.__return_code: int = 0 - os.mkdir(self.__dir_name) - - @property - def return_code(self) -> int: - return self.__return_code - - @return_code.setter - def return_code(self, return_code) -> None: - self.__return_code = return_code - if self.__return_code == 1: - if len(os.listdir(self.__dir_name)) == 0: - os.rmdir(self.__dir_name) - - def write(self, file_name: str, content: str) -> None: - with open(os.path.join(self.__dir_name, file_name), "w") as wr: - wr.write(content) - - -def open_file(file_name: str) -> int: - """Opens an already existing or - a new file in default text editor for - one of the three major os's. - - Args: - file_name: file to open. - - Returns: - Returns an integer return code indicating - if the operation was successfull or not - """ - try: - file_obj = open(file_name, "w") - except PermissionError: - return 1 - util: str = "" - cmd: List[str] = [""] - if platform.system() == "Windows": - util = "start" - elif platform.system() == "Linux": - util = "xdg-open" - file_obj.write("# Placeholder text.") - elif platform.system() == "Darwin": - util = "open" - cmd.append("-t") - file_obj.close() - cmd[0] = util - cmd.append(file_name) - ret_code: int = subprocess.run(cmd, stdout=subprocess.DEVNULL).returncode - - return ret_code - - -def gist_filename_validated(file_name: str) -> bool: - """Validates the filename. - - Validates and checks if the filename has filename - pattern prohibited by github and depending so,returns - a boolean to represent the respective state. - - Args: - file_name: filename to validate. - Returns: - boolean representing the validated state of filename. - - """ - striped_filename: str = file_name.replace("gistfile", "") - # so the logic here is that if the filename contains 'gistfile' - # in it then that will be replaced with emtpy string and the filename - # will change,otherwise it won't change which will help us decide - # if it does contain that string and if it does then the remaining - # string will be validated for numerical digits - if striped_filename != file_name and striped_filename.isdigit(): - return False - return True +# ╔═══╗ +# ║╔═╗║ +# ║║ ╚╝╔══╗╔══╗ +# ║║ ╔╗║╔╗║║╔╗║ +# ║╚═╝║║║═╣║╚╝║ +# ╚═══╝╚══╝╚═╗║ +# ╔═╝║ +# ╚══╝ +# ©Justaus3r 2022 +# This file is part of "Ceg",a gist crud utility. +# Distributed under GPLV3 +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +""" Misc stuff dat i couldn't figure out where to put """ + +import os +import platform +import subprocess +from .exceptions import CegExceptions +from typing import List, Dict, Callable, Union, Optional, Type, TypeAlias + +__all__ = ("UtilInfo", "Misc") + + +class UtilInfo: + """Metainfo about utility. + + Class containing all the meta info about + the utility. + + Attributes: + UTIL_NAME: The utility name. + DESCRIPTION: Short description of the utility. + VERSION: Semantic verson of the utility. + """ + + UTIL_NAME: str = "ceg" + UTIL_USAGE: str = f"{UTIL_NAME} [options] [sub-arguments]" + EPILOG: str = """ +sub-arguments: + --post/-po + --no-public/-np switch gist visibility to private + + --description/-desc description for the gist + + --patch/-pa + --gist-id/-gi gist-id for the gist + +For more usage help, check out https://www.github.com/justaus3r/ceg/#examples""" + DESCRIPTION: str = "An all in one github's gist manager." + # Caution(message to myself): Be careful when updating the version because + # wrong updates can be a mess. + VERSION: str = "0.5.0" + + +def exception_executioner(exception_obj) -> None: + """Raises exception taken as am argument. + + Args: + exception_obj: THe Exception object. + + """ + if exception_obj: + raise exception_obj + + +class Misc: + """Misc stuff. + + Contains all the miscellaneous vars. + + Attributes: + http_intrinsics: List containing names of all http methods. + secret_key: Gitub Secret key extracted from env variable. + """ + + http_intrinsics: List[str] = ["get", "post", "patch", "delete"] + OptionalException: TypeAlias = Optional[ + Union[ + Type[CegExceptions.BadCredentials], + Type[CegExceptions.ForbiddenResource], + Type[CegExceptions.ResourceNotFound], + Type[CegExceptions.UnprocessableRequest], + ] + ] + HttpResponseCodes: TypeAlias = Dict[ + Union[int, str], + Union[ + Dict[str, Dict[str, OptionalException]], Callable[[OptionalException], None] + ], + ] + http_response_codes: HttpResponseCodes = { + 200: {"OK!": {"exception_obj": None}}, + 201: {"Gist Created Sucessfully!": {"exception_obj": None}}, + 204: {"OK!.No Response Recieved.": {"exception_obj": None}}, + 401: {"Bad Credentials!": {"exception_obj": CegExceptions.BadCredentials}}, + 403: { + "Forbidden Resource!": {"exception_obj": CegExceptions.ForbiddenResource} + }, + 404: {"Resource not found!": {"exception_obj": CegExceptions.ResourceNotFound}}, + 422: { + "Request Unprocessable!": { + "exception_obj": CegExceptions.UnprocessableRequest + } + }, + "exception_action": exception_executioner, + } + secret_key: Optional[str] = os.getenv("GITHUB_SECRET_KEY") + + +class FileHandler: + """File & Directory Handler. + + Responsible for creating gist directories, + files,writing content to them and removing + gist directories if found empty due to some + error. + + Attributes: + return_code: return code which determines whether to keep a directory or not. + """ + + def __init__(self, dir_name: str) -> None: + "Inits FileHandler with some directory name" + self.__dir_name: str = dir_name + self.__return_code: int = 0 + os.mkdir(self.__dir_name) + + @property + def return_code(self) -> int: + return self.__return_code + + @return_code.setter + def return_code(self, return_code) -> None: + self.__return_code = return_code + if self.__return_code == 1: + if len(os.listdir(self.__dir_name)) == 0: + os.rmdir(self.__dir_name) + + def write(self, file_name: str, content: str) -> None: + with open(os.path.join(self.__dir_name, file_name), "w") as wr: + wr.write(content) + + +def open_file(file_name: str) -> int: + """Opens an already existing or + a new file in default text editor for + one of the three major os's. + + Args: + file_name: file to open. + + Returns: + Returns an integer return code indicating + if the operation was successfull or not + """ + try: + file_obj = open(file_name, "w") + except PermissionError: + return 1 + util: str = "" + cmd: List[str] = [""] + if platform.system() == "Windows": + util = "start" + elif platform.system() == "Linux": + util = "xdg-open" + file_obj.write("# Placeholder text.") + elif platform.system() == "Darwin": + util = "open" + cmd.append("-t") + file_obj.close() + cmd[0] = util + cmd.append(file_name) + ret_code: int = subprocess.run(cmd, stdout=subprocess.DEVNULL).returncode + + return ret_code + + +def gist_filename_validated(file_name: str) -> bool: + """Validates the filename. + + Validates and checks if the filename has filename + pattern prohibited by github and depending so,returns + a boolean to represent the respective state. + + Args: + file_name: filename to validate. + Returns: + boolean representing the validated state of filename. + + """ + striped_filename: str = file_name.replace("gistfile", "") + # so the logic here is that if the filename contains 'gistfile' + # in it then that will be replaced with emtpy string and the filename + # will change,otherwise it won't change which will help us decide + # if it does contain that string and if it does then the remaining + # string will be validated for numerical digits + if striped_filename != file_name and striped_filename.isdigit(): + return False + return True diff --git a/docs/ceg.html b/docs/ceg.html index 267d4e2..3f18104 100644 --- a/docs/ceg.html +++ b/docs/ceg.html @@ -3,14 +3,14 @@ - + ceg API documentation - - + +