diff --git a/.github/workflows/1.bump-version.yml b/.github/workflows/1.bump-version.yml index d7f0fae..7e99d40 100644 --- a/.github/workflows/1.bump-version.yml +++ b/.github/workflows/1.bump-version.yml @@ -25,7 +25,7 @@ jobs: - name: Install dependencies run: | python -m pip install -U pip - python -m pip install -r ./requirements/requirements.test.txt + python -m pip install .[test] - name: Test with pytest run: ./scripts/test.sh -l diff --git a/.github/workflows/2.build-publish.yml b/.github/workflows/2.build-publish.yml index 6e7529c..9684ab9 100644 --- a/.github/workflows/2.build-publish.yml +++ b/.github/workflows/2.build-publish.yml @@ -31,17 +31,17 @@ jobs: python -m pip install -U pip python -m pip install -r ./requirements/requirements.build.txt - name: Build and publish package + # run: | + # echo -e "[testpypi]\nusername = __token__\npassword = ${{ secrets.TEST_PYPI_API_TOKEN }}" > ~/.pypirc + # ./scripts/build.sh -c -u + # rm -rfv ~/.pypirc run: | - echo -e "[testpypi]\nusername = __token__\npassword = ${{ secrets.TEST_PYPI_API_TOKEN }}" > ~/.pypirc - ./scripts/build.sh -c -u + echo -e "[pypi]\nusername = __token__\npassword = ${{ secrets.PYPI_API_TOKEN }}" > ~/.pypirc + ./scripts/build.sh -c -u -p rm -rfv ~/.pypirc - # run: | - # echo -e "[pypi]\nusername = __token__\npassword = ${{ secrets.PYPI_API_TOKEN }}" > ~/.pypirc - # ./scripts/build.sh -c -u -p - # rm -rfv ~/.pypirc - # - name: Build the package - # run: | - # ./scripts/build.sh -c + - name: Build the package + run: | + ./scripts/build.sh -c - name: Create release env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/README.md b/README.md index 3297e00..bf2140d 100644 --- a/README.md +++ b/README.md @@ -4,18 +4,17 @@ [![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/bybatkhuu/module.python-utils/2.build-publish.yml?logo=GitHub)](https://github.com/bybatkhuu/module.python-utils/actions/workflows/2.build-publish.yml) [![GitHub release (latest SemVer)](https://img.shields.io/github/v/release/bybatkhuu/module.python-utils?logo=GitHub&color=blue)](https://github.com/bybatkhuu/module.python-utils/releases) -'potato_utils' is collection of useful utils package for python. +'potato_utils' is collection of simple useful utils package for python. ## ✨ Features -- Python module/package -- Configuration -- Test -- Build -- Documentation -- Scripts -- Examples -- CI/CD +- Python utilities +- Datetime utilities +- File I/O utilities +- HTTP utilities +- Security utilities +- Sanitation utilities +- Validation utilities --- @@ -75,11 +74,7 @@ git clone git@github.com:bybatkhuu/module.python-utils.git && \ **OPTION A.** [**RECOMMENDED**] Install from **PyPi**: ```sh -# Install from staging TestPyPi: -pip install -i https://test.pypi.org/simple -U potato_utils - -# Or install from production PyPi: -# pip install -U potato_utils +pip install -U potato_utils ``` **OPTION B.** Install latest version directly from **GitHub** repository: @@ -101,7 +96,7 @@ pip install -e . **OPTION D.** Install for **DEVELOPMENT** environment: ```sh -pip install -r ./requirements/requirements.dev.txt +pip install -e .[dev] ``` **OPTION E.** Install from **pre-built release** files: @@ -136,66 +131,12 @@ cp -r ./src/potato_utils /some/path/project/ [**`examples/simple/main.py`**](./examples/simple/main.py): ```python -# Standard libraries -import sys -import logging - -# Internal modules -from potato_utils import MyClass - - -logger = logging.getLogger(__name__) - - -def main() -> None: - logging.basicConfig( - stream=sys.stdout, - level=logging.INFO, - datefmt="%Y-%m-%d %H:%M:%S %z", - format="[%(asctime)s | %(levelname)s | %(filename)s:%(lineno)d]: %(message)s", - ) - - # Pre-defined variables (for customizing and testing) - _items = [0.1, 0.2, 0.3, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] - _config = { - "min_length": 4, - "max_length": 10, - "min_value": 0.0, - "max_value": 1.0, - "threshold": 0.7, - } - - # Main example code - logger.info(f"Items before cleaning: {_items}") - _my_object = MyClass(items=_items, config=_config) - _items = _my_object() - logger.info(f"Items after cleaning: {_items}") - - logger.info("Done!\n") - return - - -if __name__ == "__main__": - main() ``` 👍 --- -## ⚙️ Configuration - -[**`templates/configs/config.yml`**](./templates/configs/config.yml): - -```yaml -potato_utils: - min_length: 2 - max_length: 100 - min_value: 0.0 - max_value: 1.0 - threshold: 0.5 -``` - ### 🌎 Environment Variables [**`.env.example`**](./.env.example): @@ -214,7 +155,7 @@ To run tests, run the following command: ```sh # Install python test dependencies: -pip install -r ./requirements/requirements.test.txt +pip install .[test] # Run tests: python -m pytest -sv -o log_cli=true diff --git a/docs/.nav.yml b/docs/.nav.yml index 064174f..fcbb369 100644 --- a/docs/.nav.yml +++ b/docs/.nav.yml @@ -5,6 +5,5 @@ nav: - Development: dev/ - Research: research/ # - "*" - - Blog: blog/ - Release Notes: release-notes.md - About: about/ diff --git a/docs/README.md b/docs/README.md index 066c2d3..240e926 100644 --- a/docs/README.md +++ b/docs/README.md @@ -7,15 +7,14 @@ hide: # Introduction -'potato_utils' is collection of useful utils package for python. +'potato_utils' is collection of simple useful utils package for python. ## ✨ Features -- Python module/package -- Configuration -- Test -- Build -- Documentation -- Scripts -- Examples -- CI/CD +- Python utilities +- Datetime utilities +- File I/O utilities +- HTTP utilities +- Security utilities +- Sanitation utilities +- Validation utilities diff --git a/docs/api-docs/.nav.yml b/docs/api-docs/.nav.yml index af70428..7a6c3d5 100644 --- a/docs/api-docs/.nav.yml +++ b/docs/api-docs/.nav.yml @@ -1,3 +1,2 @@ nav: - - MyClass: MyClass.md - # - "*" + - "*" diff --git a/docs/api-docs/MyClass.md b/docs/api-docs/MyClass.md deleted file mode 100644 index 2473925..0000000 --- a/docs/api-docs/MyClass.md +++ /dev/null @@ -1,7 +0,0 @@ ---- -title: MyClass ---- - -# MyClass - -::: src.potato_utils.MyClass diff --git a/docs/api-docs/dt.md b/docs/api-docs/dt.md new file mode 100644 index 0000000..be49c91 --- /dev/null +++ b/docs/api-docs/dt.md @@ -0,0 +1,7 @@ +--- +title: Datetime Utilities +--- + +# Datetime Utilities + +::: src.potato_utils.dt diff --git a/docs/api-docs/http.md b/docs/api-docs/http.md new file mode 100644 index 0000000..9f3097c --- /dev/null +++ b/docs/api-docs/http.md @@ -0,0 +1,7 @@ +--- +title: HTTP Utilities +--- + +# HTTP Utilities + +::: src.potato_utils.http diff --git a/docs/api-docs/io.md b/docs/api-docs/io.md new file mode 100644 index 0000000..7194250 --- /dev/null +++ b/docs/api-docs/io.md @@ -0,0 +1,7 @@ +--- +title: IO Utilities +--- + +# IO Utilities + +::: src.potato_utils.io diff --git a/docs/api-docs/sanitizer.md b/docs/api-docs/sanitizer.md new file mode 100644 index 0000000..64d7d69 --- /dev/null +++ b/docs/api-docs/sanitizer.md @@ -0,0 +1,7 @@ +--- +title: Sanitizer Utilities +--- + +# Sanitizer Utilities + +::: src.potato_utils.sanitizer diff --git a/docs/api-docs/secure.md b/docs/api-docs/secure.md new file mode 100644 index 0000000..d73b8eb --- /dev/null +++ b/docs/api-docs/secure.md @@ -0,0 +1,7 @@ +--- +title: Secure Utilities +--- + +# Secure Utilities + +::: src.potato_utils.secure diff --git a/docs/api-docs/validator.md b/docs/api-docs/validator.md new file mode 100644 index 0000000..fbc17c9 --- /dev/null +++ b/docs/api-docs/validator.md @@ -0,0 +1,7 @@ +--- +title: Validator Utilities +--- + +# Validator Utilities + +::: src.potato_utils.validator diff --git a/docs/assets/images/logo.png b/docs/assets/images/logo.png index f1ff4b2..afa28ad 100644 Binary files a/docs/assets/images/logo.png and b/docs/assets/images/logo.png differ diff --git a/docs/blog/.authors.yml b/docs/blog/.authors.yml deleted file mode 100644 index 2417010..0000000 --- a/docs/blog/.authors.yml +++ /dev/null @@ -1,6 +0,0 @@ -authors: - bybatkhuu: - name: Batkhuu Byambajav - description: Creator, maintainer - avatar: https://avatars.githubusercontent.com/bybatkhuu - url: https://github.com/bybatkhuu diff --git a/docs/blog/.meta.yml b/docs/blog/.meta.yml deleted file mode 100644 index 2340187..0000000 --- a/docs/blog/.meta.yml +++ /dev/null @@ -1,2 +0,0 @@ -tags: - - blog diff --git a/docs/blog/index.md b/docs/blog/index.md deleted file mode 100644 index 684f567..0000000 --- a/docs/blog/index.md +++ /dev/null @@ -1,7 +0,0 @@ ---- -title: Blog ---- - -# ✏️ Blog - -This is the blog page. It will list all the blog posts. diff --git a/docs/blog/posts/post-1.md b/docs/blog/posts/post-1.md deleted file mode 100644 index 0d5ee33..0000000 --- a/docs/blog/posts/post-1.md +++ /dev/null @@ -1,18 +0,0 @@ ---- -date: - created: 2025-01-01T01:00:00 - updated: 2025-01-31T01:00:00 -authors: - - bybatkhuu -categories: - - Category-1 - - Category-2 -tags: - - Tag-1 - - Tag-2 -readtime: 3 ---- - -# Post 1 - -Lorem ipsum odor amet, consectetuer adipiscing elit. Mus aptent habitant ornare scelerisque vulputate ridiculus. Justo lectus massa magna duis urna. Dui elit mi potenti duis placerat efficitur dui pretium velit. Nisl cubilia fringilla eros magnis torquent? Hac proin dis ligula at sagittis sed justo volutpat facilisis. Sem netus volutpat urna cras maximus ridiculus leo. Ad vivamus tortor luctus ac ac. Consectetur nulla aptent eleifend euismod sem aliquet auctor. Nisl sem porta platea tincidunt elit tellus lacus ligula tortor. Sagittis nascetur potenti ultrices euismod, aliquet non aenean. Convallis urna gravida quisque odio cras himenaeos odio tristique. Nibh bibendum tempor leo mi ante etiam. Euismod dis praesent natoque auctor dignissim enim condimentum. Et sapien consectetur habitasse, dignissim at eleifend. Fusce laoreet efficitur sem suscipit platea tortor purus cursus magnis. diff --git a/docs/dev/sitemap.md b/docs/dev/sitemap.md index ea939c9..55dd6c2 100644 --- a/docs/dev/sitemap.md +++ b/docs/dev/sitemap.md @@ -10,9 +10,7 @@ title: Sitemap - [Installation](../getting-started/installation.md) - [Configuration](../getting-started/configuration.md) - [Examples](../getting-started/examples.md) - - [Error Codes](../getting-started/error-codes.md) - API Documentation - - [MyClass](../api-docs/MyClass.md) - Development - [Test](../dev/test.md) - [Build](../dev/build.md) @@ -43,7 +41,6 @@ title: Sitemap - [Benchmarks](../research/benchmarks.md) - [References](../research/references.md) - [Release Notes](../release-notes.md) -- [Blog](../blog/index.md) - About - [FAQ](../about/faq.md) - [Authors](../about/authors.md) diff --git a/docs/dev/test.md b/docs/dev/test.md index 7d566d6..a6515a0 100644 --- a/docs/dev/test.md +++ b/docs/dev/test.md @@ -8,7 +8,7 @@ To run tests, run the following command: ```sh # Install python test dependencies: -pip install -r ./requirements/requirements.test.txt +pip install .[test] # Run tests: python -m pytest -sv -o log_cli=true diff --git a/docs/getting-started/.nav.yml b/docs/getting-started/.nav.yml index 915bcf6..e5393c8 100644 --- a/docs/getting-started/.nav.yml +++ b/docs/getting-started/.nav.yml @@ -3,5 +3,4 @@ nav: - Installation: installation.md - Configuration: configuration.md - Examples: examples.md - - Error Codes: error-codes.md # - "*" diff --git a/docs/getting-started/configuration.md b/docs/getting-started/configuration.md index e34996c..c860325 100644 --- a/docs/getting-started/configuration.md +++ b/docs/getting-started/configuration.md @@ -2,14 +2,6 @@ title: Configuration --- -# ⚙️ Configuration - -[**`templates/configs/config.yml`**](https://github.com/bybatkhuu/module.python-utils/blob/main/templates/configs/config.yml): - -```yaml ---8<-- "./templates/configs/config.yml" -``` - ## 🌎 Environment Variables [**`.env.example`**](https://github.com/bybatkhuu/module.python-utils/blob/main/.env.example): diff --git a/docs/getting-started/error-codes.md b/docs/getting-started/error-codes.md deleted file mode 100644 index 898117a..0000000 --- a/docs/getting-started/error-codes.md +++ /dev/null @@ -1,47 +0,0 @@ ---- -title: Error Codes ---- - -# 🚨 Error Codes - -## Error Handling - -Pydantic will raise a `ValidationError` whenever it finds an error in the data it's validating. - -```python -from pydantic import ValidationError - -from potato_utils import MyClass - - -try: - _my_object = MyClass( - config={ - "min_length": 0, - "max_length": "three", - "min_value": True, - "max_value": [1, 2, 3], - "threshold": 2.0, - } - ) -except ValidationError as err: - print(err) -``` - -The error message will look like this: - -```txt -4 validation errors for MyClassConfigPM -min_length - Input should be greater than or equal to 1 [type=greater_than_equal, input_value=0, input_type=int] - For further information visit https://errors.pydantic.dev/2.10/v/greater_than_equal -max_length - Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='three', input_type=str] - For further information visit https://errors.pydantic.dev/2.10/v/int_parsing -max_value - Input should be a valid number [type=float_type, input_value=[1, 2, 3], input_type=list] - For further information visit https://errors.pydantic.dev/2.10/v/float_type -threshold - Input should be less than or equal to 1 [type=less_than_equal, input_value=2.0, input_type=float] - For further information visit https://errors.pydantic.dev/2.10/v/less_than_equal -``` diff --git a/docs/getting-started/installation.md b/docs/getting-started/installation.md index 348eac5..4958cfd 100644 --- a/docs/getting-started/installation.md +++ b/docs/getting-started/installation.md @@ -46,11 +46,7 @@ git clone git@github.com:bybatkhuu/module.python-utils.git && \ **OPTION A.** [**RECOMMENDED**] Install from **PyPi**: ```sh -# Install from staging TestPyPi: -pip install -i https://test.pypi.org/simple -U potato_utils - -# Or install from production PyPi: -# pip install -U potato_utils +pip install -U potato_utils ``` **OPTION B.** Install latest version directly from **GitHub** repository: @@ -72,7 +68,7 @@ pip install -e . **OPTION D.** Install for **DEVELOPMENT** environment: ```sh -pip install -r ./requirements/requirements.dev.txt +pip install -e .[dev] ``` **OPTION E.** Install from **pre-built release** files: diff --git a/examples/simple/main.py b/examples/simple/main.py index 2a0379b..519841f 100755 --- a/examples/simple/main.py +++ b/examples/simple/main.py @@ -5,7 +5,7 @@ import logging # Internal modules -from potato_utils import MyClass +# from potato_utils import io as io_utils logger = logging.getLogger(__name__) @@ -19,22 +19,6 @@ def main() -> None: format="[%(asctime)s | %(levelname)s | %(filename)s:%(lineno)d]: %(message)s", ) - # Pre-defined variables (for customizing and testing) - _items = [0.1, 0.2, 0.3, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] - _config = { - "min_length": 4, - "max_length": 10, - "min_value": 0.0, - "max_value": 1.0, - "threshold": 0.7, - } - - # Main example code - logger.info(f"Items before cleaning: {_items}") - _my_object = MyClass(items=_items, config=_config) - _items = _my_object.run() - logger.info(f"Items after cleaning: {_items}") - logger.info("Done!\n") return diff --git a/mkdocs.yml b/mkdocs.yml index 9b499eb..c0dfd43 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -1,5 +1,5 @@ site_name: Potato Utils (Python Utils) -site_description: "'potato_utils' is collection of useful utils package for python." +site_description: "'potato_utils' is collection of simple useful utils package for python." site_url: https://pyutils-docs.bybatkhuu.dev repo_name: bybatkhuu/module.python-utils repo_url: https://github.com/bybatkhuu/module.python-utils @@ -33,22 +33,22 @@ theme: - content.code.annotate palette: - media: "(prefers-color-scheme)" - primary: black - accent: black + primary: brown + accent: brown toggle: icon: material/brightness-auto name: Switch to light mode - media: "(prefers-color-scheme: light)" scheme: default - primary: white - accent: white + primary: brown + accent: brown toggle: icon: material/brightness-7 name: Switch to dark mode - media: "(prefers-color-scheme: dark)" scheme: slate - primary: black - accent: black + primary: brown + accent: brown toggle: icon: material/brightness-4 name: Switch to light mode @@ -97,20 +97,6 @@ plugins: - search - awesome-nav - mkdocstrings - - blog: - # post_excerpt: required - post_excerpt_max_authors: 3 - post_date_format: medium - archive_date_format: yyyy/MM - archive_url_date_format: yyyy/MM - blog_toc: true - draft: true - draft_if_future_date: true - pagination_format: "$link_first $link_previous ~2~ $link_next $link_last" - # categories_allowed: - # - Category 1 - # - Category 2 - # - Others extra: version: provider: mike diff --git a/pyproject.toml b/pyproject.toml index 3a41f09..332e283 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,10 +5,10 @@ build-backend = "setuptools.build_meta" [project] name = "potato_utils" authors = [{ name = "Batkhuu Byambajav", email = "batkhuu10@gmail.com" }] -description = "'potato_utils' is collection of useful utils package for python." +description = "'potato_utils' is collection of simple useful utils package for python." readme = "README.md" requires-python = ">=3.10,<4.0" -keywords = ["potato_utils", "template", "module", "python", "package"] +keywords = ["potato_utils", "utils", "utilities", "tools", "helpers"] license-files = ["LICEN[CS]E*"] classifiers = [ "Development Status :: 4 - Beta", @@ -31,8 +31,32 @@ dynamic = ["version", "dependencies", "optional-dependencies"] [tool.setuptools.dynamic] version = { attr = "potato_utils.__version__.__version__" } dependencies = { file = "./requirements.txt" } -optional-dependencies.extra = { file = [ - "./requirements/requirements.extra.txt", +optional-dependencies.async = { file = [ + "./requirements/requirements.async.txt", +] } +optional-dependencies.fastapi = { file = [ + "./requirements/requirements.fastapi.txt", +] } +optional-dependencies.all = { file = [ + "./requirements/requirements.async.txt", + "./requirements/requirements.fastapi.txt", +] } +optional-dependencies.test = { file = [ + "./requirements/requirements.async.txt", + "./requirements/requirements.fastapi.txt", + "./requirements/requirements.test.txt", +] } +optional-dependencies.build = { file = [ + "./requirements/requirements.build.txt", +] } +optional-dependencies.docs = { file = ["./requirements/requirements.docs.txt"] } +optional-dependencies.dev = { file = [ + "./requirements/requirements.async.txt", + "./requirements/requirements.fastapi.txt", + "./requirements/requirements.test.txt", + "./requirements/requirements.build.txt", + "./requirements/requirements.docs.txt", + "./requirements/requirements.dev.txt", ] } # [tool.pyright] diff --git a/requirements.txt b/requirements.txt index eb90e71..9b5fc74 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ python-dotenv>=1.0.1,<2.0.0 -pydantic>=2.0.3,<3.0.0 +pydantic[email,timezone]>=2.0.3,<3.0.0 pydantic-settings>=2.2.1,<3.0.0 diff --git a/requirements/requirements.async.txt b/requirements/requirements.async.txt new file mode 100644 index 0000000..67afe77 --- /dev/null +++ b/requirements/requirements.async.txt @@ -0,0 +1,3 @@ +aiofiles>=24.1.0,<25.0.0 +aioshutil>=1.5,<2.0.0 +aiohttp>=3.11.18,<4.0.0 diff --git a/requirements/requirements.dev.txt b/requirements/requirements.dev.txt index 94ca431..bf3b69f 100644 --- a/requirements/requirements.dev.txt +++ b/requirements/requirements.dev.txt @@ -1,7 +1,6 @@ --e . --r ./requirements.test.txt --r ./requirements.build.txt --r ./requirements.docs.txt --r ./requirements.extra.txt +# -e .[all] +# -r ./requirements.test.txt +# -r ./requirements.build.txt +# -r ./requirements.docs.txt pyright>=1.1.392,<2.0.0 pre-commit>=4.0.1,<5.0.0 diff --git a/requirements/requirements.extra.txt b/requirements/requirements.extra.txt deleted file mode 100644 index 7a87be5..0000000 --- a/requirements/requirements.extra.txt +++ /dev/null @@ -1 +0,0 @@ -pydantic[email,timezone]>=2.0.3,<3.0.0 diff --git a/requirements/requirements.fastapi.txt b/requirements/requirements.fastapi.txt new file mode 100644 index 0000000..1c39506 --- /dev/null +++ b/requirements/requirements.fastapi.txt @@ -0,0 +1 @@ +fastapi>=0.109.2,<1.0.0 diff --git a/requirements/requirements.test.txt b/requirements/requirements.test.txt index a884daf..64ab717 100644 --- a/requirements/requirements.test.txt +++ b/requirements/requirements.test.txt @@ -1,4 +1,4 @@ --r ../requirements.txt +# -r ../requirements.txt pytest>=8.0.2,<9.0.0 pytest-cov>=5.0.0,<8.0.0 pytest-xdist>=3.6.1,<4.0.0 diff --git a/src/potato_utils/__init__.py b/src/potato_utils/__init__.py index d11afa2..2d28854 100644 --- a/src/potato_utils/__init__.py +++ b/src/potato_utils/__init__.py @@ -1,11 +1,6 @@ from .__version__ import __version__ -from .config import MyClassConfigPM, MyClassCliConfig -from ._base import MyClass __all__ = [ "__version__", - "MyClassConfigPM", - "MyClassCliConfig", - "MyClass", ] diff --git a/src/potato_utils/__main__.py b/src/potato_utils/__main__.py deleted file mode 100644 index 3c68c00..0000000 --- a/src/potato_utils/__main__.py +++ /dev/null @@ -1,39 +0,0 @@ -# Standard libraries -import os -import sys -import logging - -# Third-party libraries -from dotenv import load_dotenv - -# Internal modules -from .config import MyClassCliConfig -from ._base import MyClass - - -load_dotenv(dotenv_path=".env", override=True) -logger = logging.getLogger(__name__) - - -def main() -> None: - _log_level = logging.INFO - if str(os.getenv("DEBUG", "0")).lower() in ("1", "true", "t", "yes", "y"): - _log_level = logging.DEBUG - - logging.basicConfig( - stream=sys.stdout, - level=_log_level, - datefmt="%Y-%m-%d %H:%M:%S %z", - format="[%(asctime)s | %(levelname)s | %(filename)s:%(lineno)d]: %(message)s", - ) - - _my_class_config = MyClassCliConfig() # type: ignore - _my_object = MyClass(items=_my_class_config.items, config=_my_class_config) - _items = _my_object.run() - logger.info(f"Items: {_items}") - - return - - -if __name__ == "__main__": - main() diff --git a/src/potato_utils/_base.py b/src/potato_utils/_base.py index c720ce6..eeaadc8 100644 --- a/src/potato_utils/_base.py +++ b/src/potato_utils/_base.py @@ -1,190 +1,117 @@ -# Standard libraries -import pprint +import re +import copy import logging -from typing import Any -# Third-party libraries from pydantic import validate_call -# Internal modules -from .__version__ import __version__ -from . import _utils as utils -from .config import MyClassConfigPM - logger = logging.getLogger(__name__) -class MyClass: - """A class to perform some operations on a list of float items. - - Attributes: - items (list[float] ): List of float items to be processed. - config (MyClassConfigPM): Configuration for the module. +@validate_call +def deep_merge(dict1: dict, dict2: dict) -> dict: + """Return a new dictionary that's the result of a deep merge of two dictionaries. + If there are conflicts, values from `dict2` will overwrite those in `dict1`. - Methods: - run(): Method to clean the items based on the threshold value. + Args: + dict1 (dict, required): The base dictionary that will be merged. + dict2 (dict, required): The dictionary to merge into `dict1`. + Returns: + dict: The merged dictionary. """ - @validate_call - def __init__( - self, - items: list[float] | None = None, - config: MyClassConfigPM | dict[str, Any] | None = None, - auto_run: bool = False, - **kwargs, - ) -> None: - """Initializer method for the MyClass class. - - Args: - items (list[float] | None , optional): List of float items to be processed. - Defaults to None. - config (MyClassConfigPM | dict[str, Any] | None, optional): Configuration for the module. Defaults to None. - """ - - logger.debug( - f"Initializing <{self.__class__.__name__}> object with '{__version__}' version..." - ) - if not config: - config = MyClassConfigPM() - - self.config = config - if kwargs: - self.config = self.config.model_copy(update=kwargs) - - if items: - self.items = items - logger.debug( - f"Initialized <{self.__class__.__name__}> object with '{__version__}' version." - ) - - if auto_run: - self.run() - - @validate_call - def run( - self, - items: list[float] | None = None, - threshold: float | None = None, - ) -> list[float]: - """Method to clean the items based on the threshold value. - - Args: - items (list[float] | None, optional): List of float items to be processed. Defaults to None. - threshold (float | None , optional): Threshold value for the cleaning process. Defaults to None. - - Raises: - RuntimeError: If `items` attribute is not set. - - Returns: - list[float]: List of cleaned items. - """ - - if items: - self.items = items - - if not hasattr(self, "items"): - raise RuntimeError( - "`items` attribute is not set, must provide a list of float items to be processed!" - ) - - if not threshold: - threshold = self.config.threshold - - logger.debug(f"Cleaning items with threshold '{threshold}'...") - _clean_items = [] - for _item in self.items: - if threshold <= _item: - _clean_items.append(_item) - else: - logger.debug( - f"Item '{_item}' is below the threshold '{threshold}', removing it..." - ) - - logger.debug("Successfully cleaned items.") - - self.items = _clean_items - return self.items - - # ATTRIBUTES - # config - @property - def config(self) -> MyClassConfigPM: - try: - return self.__config - except AttributeError: - self.__config = MyClassConfigPM() - - return self.__config - - @config.setter - def config(self, config: MyClassConfigPM | dict[str, Any]) -> None: - if (not isinstance(config, MyClassConfigPM)) and (not isinstance(config, dict)): - raise TypeError( - f"`config` attribute type {type(config)} is invalid, must be a or !" - ) - - if isinstance(config, dict): - config = MyClassConfigPM(**config) - elif isinstance(config, MyClassConfigPM): - config = config.model_copy(deep=True) - - self.__config = config - - # config - - # items - @property - def items(self) -> list[float]: - try: - return self.__items - except AttributeError: - raise AttributeError("`items` attribute is not set!") - - @items.setter - def items(self, items: list[float]) -> None: - if not isinstance(items, list): - raise TypeError( - f"`items` attribute type {type(items)} is invalid, must be a !" - ) - - if (len(items) < self.config.min_length) or ( - self.config.max_length < len(items) + _merged = copy.deepcopy(dict1) + for _key, _val in dict2.items(): + if ( + _key in _merged + and isinstance(_merged[_key], dict) + and isinstance(_val, dict) ): - raise ValueError( - f"`items` attribute length '{len(items)}' is too short or too long, " - f"must be between '{self.config.min_length}' and '{self.config.max_length}'!" - ) + _merged[_key] = deep_merge(_merged[_key], _val) + else: + _merged[_key] = copy.deepcopy(_val) + + return _merged + + +@validate_call +def camel_to_snake(val: str) -> str: + """Convert CamelCase to snake_case. + + Args: + val (str): CamelCase string to convert. + + Returns: + str: Converted snake_case string. + """ + + val = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", val) + val = re.sub("([a-z0-9])([A-Z])", r"\1_\2", val).lower() + return val - for _item in items: - if not isinstance(_item, float): - raise TypeError( - f"`items` attribute item type {type(_item)} is invalid, must be a !" - ) - if (_item < self.config.min_value) or (self.config.max_value < _item): - raise ValueError( - f"`items` attribute item value '{_item}' is not in the allowed range, " - f"must be between '{self.config.min_value}' and '{self.config.max_value}'!" - ) +@validate_call +def clean_obj_dict(obj_dict: dict, cls_name: str) -> dict: + """Clean class name from object.__dict__ for str(object). - self.__items = items + Args: + obj_dict (dict, required): Object dictionary by object.__dict__. + cls_name (str , required): Class name by cls.__name__. + + Returns: + dict: Clean object dictionary. + """ - # items - # ATTRIBUTES + try: + if not obj_dict: + raise ValueError("'obj_dict' argument value is empty!") - # METHOD OVERRIDING - def __str__(self): - _self_dict = utils.clean_obj_dict(self.__dict__, self.__class__.__name__) - _self_str = f"{self.__class__.__name__}: {pprint.pformat(_self_dict)}" - return _self_str + if not cls_name: + raise ValueError("'cls_name' argument value is empty!") + except ValueError as err: + logger.error(err) + raise - def __repr__(self): - _self_repr = utils.obj_to_repr(self) - return _self_repr + _self_dict = obj_dict.copy() + for _key in _self_dict.copy(): + _class_prefix = f"_{cls_name}__" + if _key.startswith(_class_prefix): + _new_key = _key.replace(_class_prefix, "") + _self_dict[_new_key] = _self_dict.pop(_key) + return _self_dict - # METHOD OVERRIDING +@validate_call(config={"arbitrary_types_allowed": True}) +def obj_to_repr(obj: object) -> str: + """Modifying object default repr() to custom info. + + Args: + obj (object, required): Any python object. + + Returns: + str: String for repr() method. + """ -__all__ = ["MyClass"] + try: + if not obj: + raise ValueError("'obj' argument value is empty!") + except ValueError as err: + logger.error(err) + raise + + _self_repr = ( + f"<{obj.__class__.__module__}.{obj.__class__.__name__} object at {hex(id(obj))}: " + + "{" + + f"{str(dir(obj)).replace('[', '').replace(']', '')}" + + "}>" + ) + return _self_repr + + +__all__ = [ + "deep_merge", + "camel_to_snake", + "clean_obj_dict", + "obj_to_repr", +] diff --git a/src/potato_utils/_utils.py b/src/potato_utils/_utils.py deleted file mode 100644 index 160536d..0000000 --- a/src/potato_utils/_utils.py +++ /dev/null @@ -1,72 +0,0 @@ -import logging - -from pydantic import validate_call - - -logger = logging.getLogger(__name__) - - -@validate_call -def clean_obj_dict(obj_dict: dict, cls_name: str) -> dict: - """Clean class name from object.__dict__ for str(object). - - Args: - obj_dict (dict): Object dictionary by object.__dict__. - cls_name (str ): Class name by cls.__name__. - - Returns: - dict: Clean object dictionary. - """ - - try: - if not obj_dict: - raise ValueError("'obj_dict' argument value is empty!") - - if not cls_name: - raise ValueError("'cls_name' argument value is empty!") - except ValueError as err: - logger.error(err) - raise - - _self_dict = obj_dict.copy() - for _key in _self_dict.copy(): - _class_prefix = f"_{cls_name}__" - if _key.startswith(_class_prefix): - _new_key = _key.replace(_class_prefix, "") - _self_dict[_new_key] = _self_dict.pop(_key) - - return _self_dict - - -@validate_call(config={"arbitrary_types_allowed": True}) -def obj_to_repr(obj: object) -> str: - """Modifying object default repr() to custom info. - - Args: - obj (object): Any python object. - - Returns: - str: String for repr() method. - """ - - try: - if not obj: - raise ValueError("'obj' argument value is empty!") - - except ValueError as err: - logger.error(err) - raise - - _self_repr = ( - f"<{obj.__class__.__module__}.{obj.__class__.__name__} object at {hex(id(obj))}: " - + "{" - + f"{str(dir(obj)).replace('[', '').replace(']', '')}" - + "}>" - ) - return _self_repr - - -__all__ = [ - "clean_obj_dict", - "obj_to_repr", -] diff --git a/src/potato_utils/config.py b/src/potato_utils/config.py deleted file mode 100644 index ce6772d..0000000 --- a/src/potato_utils/config.py +++ /dev/null @@ -1,62 +0,0 @@ -from pydantic import BaseModel, Field, ConfigDict -from pydantic_settings import ( - BaseSettings, - SettingsConfigDict, - PydanticBaseSettingsSource, -) - - -# Pydantic Model Config -class MyClassConfigPM(BaseModel): - min_length: int = Field( - default=2, ge=1, le=1000, description="Minimum length of the list." - ) - max_length: int = Field( - default=100, ge=1, le=1000, description="Maximum length of the list." - ) - min_value: float = Field( - default=0.0, ge=0.0, le=1.0, description="Minimum value of the each item." - ) - max_value: float = Field( - default=1.0, ge=0.0, le=1.0, description="Maximum value of the each item." - ) - threshold: float = Field( - default=0.05, ge=0.0, le=1.0, description="Threshold value to filter items." - ) - - model_config = ConfigDict(extra="allow") - - -ENV_PREFIX = "PU_" - - -# Pydantic Cli Config -class MyClassCliConfig(MyClassConfigPM, BaseSettings): - - items: list[float] = Field(..., description="List of float items to be cleaned.") - - @classmethod - def settings_customise_sources( - cls, - settings_cls: type[BaseSettings], - init_settings: PydanticBaseSettingsSource, - env_settings: PydanticBaseSettingsSource, - dotenv_settings: PydanticBaseSettingsSource, - file_secret_settings: PydanticBaseSettingsSource, - ) -> tuple[PydanticBaseSettingsSource, ...]: - return dotenv_settings, env_settings, file_secret_settings, init_settings - - model_config = SettingsConfigDict( - env_file=".env", - env_prefix=ENV_PREFIX, - env_nested_delimiter="__", - cli_parse_args=True, - cli_enforce_required=True, - ) - - -__all__ = [ - "ENV_PREFIX", - "MyClassConfigPM", - "MyClassCliConfig", -] diff --git a/src/potato_utils/constants/__init__.py b/src/potato_utils/constants/__init__.py new file mode 100644 index 0000000..26ed4bc --- /dev/null +++ b/src/potato_utils/constants/__init__.py @@ -0,0 +1,5 @@ +# flake8: noqa + +from ._base import * +from ._enum import * +from ._regex import * diff --git a/src/potato_utils/constants/_base.py b/src/potato_utils/constants/_base.py new file mode 100644 index 0000000..663d58a --- /dev/null +++ b/src/potato_utils/constants/_base.py @@ -0,0 +1,5 @@ +MAX_PATH_LENGTH = 1024 + +__all__ = [ + "MAX_PATH_LENGTH", +] diff --git a/src/potato_utils/constants/_enum.py b/src/potato_utils/constants/_enum.py new file mode 100644 index 0000000..7020761 --- /dev/null +++ b/src/potato_utils/constants/_enum.py @@ -0,0 +1,31 @@ +from enum import Enum + + +class WarnEnum(str, Enum): + ERROR = "ERROR" + ALWAYS = "ALWAYS" + DEBUG = "DEBUG" + IGNORE = "IGNORE" + + +class TSUnitEnum(str, Enum): + SECONDS = "SECONDS" + MILLISECONDS = "MILLISECONDS" + MICROSECONDS = "MICROSECONDS" + NANOSECONDS = "NANOSECONDS" + + +class HashAlgoEnum(str, Enum): + md5 = "md5" + sha1 = "sha1" + sha224 = "sha224" + sha256 = "sha256" + sha384 = "sha384" + sha512 = "sha512" + + +__all__ = [ + "WarnEnum", + "TSUnitEnum", + "HashAlgoEnum", +] diff --git a/src/potato_utils/constants/_regex.py b/src/potato_utils/constants/_regex.py new file mode 100644 index 0000000..1726c79 --- /dev/null +++ b/src/potato_utils/constants/_regex.py @@ -0,0 +1,23 @@ +REQUEST_ID_REGEX = ( + r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b|" + r"\b[0-9a-fA-F]{32}\b" +) + +# Invalid characters: +SPECIAL_CHARS_REGEX = r"[&'\"<>]" +SPECIAL_CHARS_BASE_REGEX = r"[&'\"<>\\\/]" +SPECIAL_CHARS_LOW_REGEX = r"[&'\"<>\\\/`{}|]" +SPECIAL_CHARS_MEDIUM_REGEX = r"[&'\"<>\\\/`{}|()\[\]]" +SPECIAL_CHARS_HIGH_REGEX = r"[&'\"<>\\\/`{}|()\[\]!@#$%^*;:?]" +SPECIAL_CHARS_STRICT_REGEX = r"[&'\"<>\\\/`{}|()\[\]~!@#$%^*_=\-+;:,.?\t\n ]" + + +__all__ = [ + "REQUEST_ID_REGEX", + "SPECIAL_CHARS_REGEX", + "SPECIAL_CHARS_BASE_REGEX", + "SPECIAL_CHARS_LOW_REGEX", + "SPECIAL_CHARS_MEDIUM_REGEX", + "SPECIAL_CHARS_HIGH_REGEX", + "SPECIAL_CHARS_STRICT_REGEX", +] diff --git a/src/potato_utils/dt.py b/src/potato_utils/dt.py new file mode 100644 index 0000000..b655114 --- /dev/null +++ b/src/potato_utils/dt.py @@ -0,0 +1,240 @@ +import time +import logging +from zoneinfo import ZoneInfo +from datetime import datetime, timezone, tzinfo, timedelta + +from pydantic import validate_call + +from .constants import WarnEnum, TSUnitEnum + + +logger = logging.getLogger(__name__) + + +@validate_call(config={"arbitrary_types_allowed": True}) +def add_tzinfo(dt: datetime, tz: ZoneInfo | tzinfo | str) -> datetime: + """Add or replace timezone info to datetime object. + + Args: + dt (datetime , required): Datetime object. + tz (Union[ZoneInfo, tzinfo, str], required): Timezone info. + + Returns: + datetime: Datetime object with timezone info. + """ + + if isinstance(tz, str): + tz = ZoneInfo(tz) + + dt = dt.replace(tzinfo=tz) + return dt + + +@validate_call +def datetime_to_iso( + dt: datetime, sep: str = "T", warn_mode: WarnEnum = WarnEnum.IGNORE +) -> str: + """Convert datetime object to ISO 8601 format. + + Args: + dt (datetime, required): Datetime object. + sep (str , optional): Separator between date and time. Defaults to "T". + warn_mode (WarnEnum, optional): Warning mode. Defaults to WarnEnum.IGNORE. + + Raises: + ValueError: If `sep` argument length is greater than 8. + ValueError: If `dt` argument doesn't have any timezone info and `warn_mode` is set to WarnEnum.ERROR. + + Returns: + str: Datetime string in ISO 8601 format. + """ + + sep = sep.strip() + if 8 < len(sep): + raise ValueError( + f"`sep` argument length '{len(sep)}' is too long, must be less than or equal to 8!" + ) + + if not dt.tzinfo: + _message = "Not found any timezone info in `dt` argument, assuming it's UTC timezone..." + if warn_mode == WarnEnum.ALWAYS: + logger.warning(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + elif warn_mode == WarnEnum.ERROR: + _message = "Not found any timezone info in `dt` argument!" + logger.error(_message) + raise ValueError(_message) + + dt = add_tzinfo(dt=dt, tz="UTC") + + _dt_str = dt.isoformat(sep=sep, timespec="milliseconds") + return _dt_str + + +@validate_call(config={"arbitrary_types_allowed": True}) +def convert_tz( + dt: datetime, tz: ZoneInfo | tzinfo | str, warn_mode: WarnEnum = WarnEnum.ALWAYS +) -> datetime: + """Convert datetime object to another timezone. + + Args: + dt (datetime , required): Datetime object to convert. + tz (Union[ZoneInfo, tzinfo, str], required): Timezone info to convert. + warn_mode (WarnEnum , optional): Warning mode. Defaults to WarnEnum.ALWAYS. + + Raises: + ValueError: If `dt` argument doesn't have any timezone info and `warn_mode` is set to WarnEnum.ERROR. + + Returns: + datetime: Datetime object which has been converted to another timezone. + """ + + if not dt.tzinfo: + _message = "Not found any timezone info in `dt` argument, assuming it's UTC timezone..." + if warn_mode == WarnEnum.ALWAYS: + logger.warning(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + elif warn_mode == WarnEnum.ERROR: + _message = "Not found any timezone info in `dt` argument!" + logger.error(_message) + raise ValueError(_message) + + dt = add_tzinfo(dt=dt, tz="UTC") + + if isinstance(tz, str): + tz = ZoneInfo(tz) + + dt = dt.astimezone(tz=tz) + return dt + + +def now_utc_dt() -> datetime: + """Get current datetime in UTC timezone with tzinfo. + + Returns: + datetime: Current datetime in UTC timezone with tzinfo. + """ + + _utc_dt = datetime.now(tz=timezone.utc) + return _utc_dt + + +def now_local_dt() -> datetime: + """Get current datetime in local timezone with tzinfo. + + Returns: + datetime: Current datetime in local timezone with tzinfo. + """ + + _local_dt = datetime.now().astimezone() + return _local_dt + + +@validate_call(config={"arbitrary_types_allowed": True}) +def now_dt(tz: ZoneInfo | tzinfo | str) -> datetime: + """Get current datetime in specified timezone with tzinfo. + + Args: + tz (Union[ZoneInfo, tzinfo, str], required): Timezone info. + + Returns: + datetime: Current datetime in specified timezone with tzinfo. + """ + + _dt = now_utc_dt() + _dt = convert_tz(dt=_dt, tz=tz) + return _dt + + +@validate_call +def now_ts(unit: TSUnitEnum = TSUnitEnum.SECONDS) -> int: + """Get current timestamp in UTC timezone. + + Args: + unit (TSUnitEnum, optional): Type of timestamp unit. Defaults to `TSUnitEnum.SECONDS`. + + Returns: + int: Current timestamp. + """ + + _now_ts: int + if unit == TSUnitEnum.SECONDS: + _now_ts = int(time.time()) + elif unit == TSUnitEnum.MILLISECONDS: + _now_ts = int(time.time() * 1000) + elif unit == TSUnitEnum.MICROSECONDS: + _now_ts = int(time.time_ns() / 1000) + elif unit == TSUnitEnum.NANOSECONDS: + _now_ts = int(time.time_ns()) + + return _now_ts + + +@validate_call +def convert_ts(dt: datetime, unit: TSUnitEnum = TSUnitEnum.SECONDS) -> int: + """Convert datetime to timestamp. + + Args: + dt (datetime , required): Datetime object to convert. + unit (TSUnitEnum, optional): Type of timestamp unit. Defaults to `TSUnitEnum.SECONDS`. + + Returns: + int: Converted timestamp. + """ + + _ts: int + if unit == TSUnitEnum.SECONDS: + _ts = int(dt.timestamp()) + elif unit == TSUnitEnum.MILLISECONDS: + _ts = int(dt.timestamp() * 1000) + elif unit == TSUnitEnum.MICROSECONDS: + _ts = int(dt.timestamp() * 1000000) + elif unit == TSUnitEnum.NANOSECONDS: + _ts = int(dt.timestamp() * 1000000000) + + return _ts + + +@validate_call(config={"arbitrary_types_allowed": True}) +def calc_future_dt( + delta: timedelta | int, + dt: datetime | None = None, + tz: ZoneInfo | tzinfo | str | None = None, +) -> datetime: + """Calculate future datetime by adding delta time to current or specified datetime. + + Args: + delta (Union[timedelta, int] , required): Delta time to add to current or specified datetime. + dt (Optional[datetime] , optional): Datetime before adding delta time. Defaults to None. + tz (Union[ZoneInfo, tzinfo, str, None], optional): Timezone info. Defaults to None. + + Returns: + datetime: Calculated future datetime. + """ + + if not dt: + dt = now_utc_dt() + + if tz: + dt = convert_tz(dt=dt, tz=tz) + + if isinstance(delta, int): + delta = timedelta(seconds=delta) + + _future_dt = dt + delta + return _future_dt + + +__all__ = [ + "add_tzinfo", + "datetime_to_iso", + "convert_tz", + "now_utc_dt", + "now_local_dt", + "now_dt", + "now_ts", + "convert_ts", + "calc_future_dt", +] diff --git a/src/potato_utils/http/__init__.py b/src/potato_utils/http/__init__.py new file mode 100644 index 0000000..e01bcfc --- /dev/null +++ b/src/potato_utils/http/__init__.py @@ -0,0 +1,12 @@ +# flake8: noqa + +import importlib.util + +from ._base import * +from ._sync import * + +_async_package_name = "aiohttp" +_async_spec = importlib.util.find_spec(_async_package_name) + +if _async_spec is not None: + from ._async import * diff --git a/src/potato_utils/http/_async.py b/src/potato_utils/http/_async.py new file mode 100644 index 0000000..1d391b2 --- /dev/null +++ b/src/potato_utils/http/_async.py @@ -0,0 +1,42 @@ +import aiohttp +from pydantic import validate_call, AnyHttpUrl + + +@validate_call +async def async_is_connectable( + url: AnyHttpUrl = AnyHttpUrl("https://www.google.com"), + timeout: int = 3, + check_status: bool = False, +) -> bool: + """Check if the url is connectable. + + Args: + url (AnyHttpUrl, optional): URL to check. Defaults to 'https://www.google.com'. + timeout (int , optional): Timeout in seconds. Defaults to 3. + check_status (bool , optional): Check HTTP status code (200). Defaults to False. + + Raise: + ValueError: If `timeout` is less than 1. + + Returns: + bool: True if connectable, False otherwise. + """ + + if timeout < 1: + raise ValueError( + f"`timeout` argument value {timeout} is invalid, must be greater than 0!" + ) + + try: + async with aiohttp.ClientSession() as _session: + async with _session.get(str(url), timeout=timeout) as _response: + if check_status: + return _response.status == 200 + return True + except Exception: + return False + + +__all__ = [ + "async_is_connectable", +] diff --git a/src/potato_utils/http/_base.py b/src/potato_utils/http/_base.py new file mode 100644 index 0000000..3cd51fa --- /dev/null +++ b/src/potato_utils/http/_base.py @@ -0,0 +1,46 @@ +from http import HTTPStatus + +from pydantic import validate_call + + +@validate_call +def get_http_status(status_code: int) -> tuple[HTTPStatus, bool]: + """Get HTTP status code enum from integer value. + + Args: + status_code (int, required): Status code for HTTP response: [100 <= status_code <= 599]. + + Raises: + ValueError: If status code is not in range [100 <= status_code <= 599]. + + Returns: + Tuple[HTTPStatus, bool]: Tuple of HTTP status code enum and boolean value if status code is known. + """ + + _http_status: HTTPStatus + _is_known_status = False + try: + _http_status = HTTPStatus(status_code) + _is_known_status = True + except ValueError: + if (100 <= status_code) and (status_code < 200): + status_code = 100 + elif (200 <= status_code) and (status_code < 300): + status_code = 200 + elif (300 <= status_code) and (status_code < 400): + status_code = 304 + elif (400 <= status_code) and (status_code < 500): + status_code = 400 + elif (500 <= status_code) and (status_code < 600): + status_code = 500 + else: + raise ValueError(f"Invalid HTTP status code: '{status_code}'!") + + _http_status = HTTPStatus(status_code) + + return (_http_status, _is_known_status) + + +__all__ = [ + "get_http_status", +] diff --git a/src/potato_utils/http/_sync.py b/src/potato_utils/http/_sync.py new file mode 100644 index 0000000..93480a9 --- /dev/null +++ b/src/potato_utils/http/_sync.py @@ -0,0 +1,45 @@ +from urllib import request +from http.client import HTTPResponse + +from pydantic import validate_call, AnyHttpUrl + + +@validate_call +def is_connectable( + url: AnyHttpUrl = AnyHttpUrl("https://www.google.com"), + timeout: int = 3, + check_status: bool = False, +) -> bool: + """Check if the url is connectable. + + Args: + url (AnyHttpUrl, optional): URL to check. Defaults to 'https://www.google.com'. + timeout (int , optional): Timeout in seconds. Defaults to 3. + check_status (bool , optional): Check HTTP status code (200). Defaults to False. + + Raise: + ValueError: If `timeout` is less than 1. + + Returns: + bool: True if connectable, False otherwise. + """ + + if timeout < 1: + raise ValueError( + f"`timeout` argument value {timeout} is invalid, must be greater than 0!" + ) + + try: + _response: HTTPResponse = request.urlopen( + str(url), timeout=timeout + ) # nosec B310 + if check_status: + return _response.getcode() == 200 + return True + except Exception: + return False + + +__all__ = [ + "is_connectable", +] diff --git a/src/potato_utils/http/fastapi.py b/src/potato_utils/http/fastapi.py new file mode 100644 index 0000000..15a3ddb --- /dev/null +++ b/src/potato_utils/http/fastapi.py @@ -0,0 +1,26 @@ +from pydantic import validate_call +from starlette.datastructures import URL +from fastapi import Request + + +@validate_call(config={"arbitrary_types_allowed": True}) +def get_relative_url(val: Request | URL) -> str: + """Get relative url only path with query params from request object or URL object. + + Args: + val (Union[Request, URL]): Request object or URL object to extract relative url. + + Returns: + str: Relative url only path with query params. + """ + + if isinstance(val, Request): + val = val.url + + _relative_url = str(val).replace(f"{val.scheme}://{val.netloc}", "") + return _relative_url + + +__all__ = [ + "get_relative_url", +] diff --git a/src/potato_utils/io/__init__.py b/src/potato_utils/io/__init__.py new file mode 100644 index 0000000..e843e61 --- /dev/null +++ b/src/potato_utils/io/__init__.py @@ -0,0 +1,11 @@ +# flake8: noqa + +import importlib.util + +from ._sync import * + +_async_package_name = "aiofiles" +_async_spec = importlib.util.find_spec(_async_package_name) + +if _async_spec is not None: + from ._async import * diff --git a/src/potato_utils/io/_async.py b/src/potato_utils/io/_async.py new file mode 100644 index 0000000..9263310 --- /dev/null +++ b/src/potato_utils/io/_async.py @@ -0,0 +1,274 @@ +import errno +import hashlib +import logging + +import aioshutil +import aiofiles.os +from pydantic import validate_call + +from ..constants import WarnEnum, HashAlgoEnum, MAX_PATH_LENGTH + + +logger = logging.getLogger(__name__) + + +@validate_call +async def async_create_dir( + create_dir: str, warn_mode: WarnEnum = WarnEnum.DEBUG +) -> None: + """Asynchronous create directory if `create_dir` doesn't exist. + + Args: + create_dir (str, required): Create directory path. + warn_mode (str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `create_dir` argument length is out of range. + OSError : When warning mode is set to ERROR and directory already exists. + OSError : If failed to create directory. + """ + + create_dir = create_dir.strip() + if (len(create_dir) < 1) or (MAX_PATH_LENGTH < len(create_dir)): + raise ValueError( + f"`create_dir` argument length {len(create_dir)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if not await aiofiles.os.path.isdir(create_dir): + try: + _message = f"Creating '{create_dir}' directory..." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + await aiofiles.os.makedirs(create_dir) + except OSError as err: + if (err.errno == errno.EEXIST) and (warn_mode == WarnEnum.DEBUG): + logger.debug(f"'{create_dir}' directory already exists!") + else: + logger.error(f"Failed to create '{create_dir}' directory!") + raise + + _message = f"Successfully created '{create_dir}' directory." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.EEXIST, f"'{create_dir}' directory already exists!") + + return + + +@validate_call +async def async_remove_dir( + remove_dir: str, warn_mode: WarnEnum = WarnEnum.DEBUG +) -> None: + """Asynchronous remove directory if `remove_dir` exists. + + Args: + remove_dir (str, required): Remove directory path. + warn_mode (str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `remove_dir` argument length is out of range. + OSError : When warning mode is set to ERROR and directory doesn't exist. + OSError : If failed to remove directory. + """ + + remove_dir = remove_dir.strip() + if (len(remove_dir) < 1) or (MAX_PATH_LENGTH < len(remove_dir)): + raise ValueError( + f"`remove_dir` argument length {len(remove_dir)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if await aiofiles.os.path.isdir(remove_dir): + try: + _message = f"Removing '{remove_dir}' directory..." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + await aioshutil.rmtree(remove_dir) + except OSError as err: + if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG): + logger.debug(f"'{remove_dir}' directory doesn't exist!") + else: + logger.error(f"Failed to remove '{remove_dir}' directory!") + raise + + _message = f"Successfully removed '{remove_dir}' directory." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.ENOENT, f"'{remove_dir}' directory doesn't exist!") + + return + + +@validate_call +async def async_remove_dirs( + remove_dirs: list[str], warn_mode: WarnEnum = WarnEnum.DEBUG +) -> None: + """Asynchronous remove directories if `remove_dirs` exists. + + Args: + remove_dirs (List[str], required): Remove directories paths as list. + warn_mode (str , optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + """ + + for _remove_dir in remove_dirs: + await async_remove_dir(remove_dir=_remove_dir, warn_mode=warn_mode) + + return + + +@validate_call +async def async_remove_file( + file_path: str, warn_mode: WarnEnum = WarnEnum.DEBUG +) -> None: + """Asynchronous remove file if `file_path` exists. + + Args: + file_path (str, required): Remove file path. + warn_mode (str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `file_path` argument length is out of range. + OSError : When warning mode is set to ERROR and file doesn't exist. + OSError : If failed to remove file. + """ + + file_path = file_path.strip() + if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)): + raise ValueError( + f"`file_path` argument length {len(file_path)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if await aiofiles.os.path.isfile(file_path): + try: + _message = f"Removing '{file_path}' file..." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + await aiofiles.os.remove(file_path) + except OSError as err: + if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG): + logger.debug(f"'{file_path}' file doesn't exist!") + else: + logger.error(f"Failed to remove '{file_path}' file!") + raise + + _message = f"Successfully removed '{file_path}' file." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.ENOENT, f"'{file_path}' file doesn't exist!") + + return + + +@validate_call +async def async_remove_files( + file_paths: list[str], warn_mode: WarnEnum = WarnEnum.DEBUG +) -> None: + """Asynchronous remove files if `file_paths` exists. + + Args: + file_paths (List[str], required): Remove file paths as list. + warn_mode (str , optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + """ + + for _file_path in file_paths: + await async_remove_file(file_path=_file_path, warn_mode=warn_mode) + + return + + +@validate_call +async def async_get_file_checksum( + file_path: str, + hash_method: HashAlgoEnum = HashAlgoEnum.md5, + chunk_size: int = 4096, + warn_mode: WarnEnum = WarnEnum.DEBUG, +) -> str | None: + """Asynchronous get file checksum. + + Args: + file_path (str , required): Target file path. + hash_method (HashAlgoEnum, optional): Hash method. Defaults to `HashAlgoEnum.md5`. + chunk_size (int , optional): Chunk size. Defaults to 4096. + warn_mode (str , optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `file_path` argument length is out of range. + ValueError: If `chunk_size` argument value is invalid. + OSError : When warning mode is set to ERROR and file doesn't exist. + + Returns: + str | None: File checksum or None if file doesn't exist. + """ + + file_path = file_path.strip() + if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)): + raise ValueError( + f"`file_path` argument length {len(file_path)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if chunk_size < 10: + raise ValueError( + f"`chunk_size` argument value {chunk_size} is invalid, must be greater than 10!" + ) + + _file_checksum: str | None = None + if await aiofiles.os.path.isfile(file_path): + _file_hash = hashlib.new(hash_method.value) + async with aiofiles.open(file_path, "rb") as _file: + while True: + _file_chunk = await _file.read(chunk_size) + if not _file_chunk: + break + _file_hash.update(_file_chunk) + + _file_checksum = _file_hash.hexdigest() + else: + _message = f"'{file_path}' file doesn't exist!" + if warn_mode == WarnEnum.ALWAYS: + logger.warning(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.ENOENT, _message) + + return _file_checksum + + +__all__ = [ + "async_create_dir", + "async_remove_dir", + "async_remove_dirs", + "async_remove_file", + "async_remove_files", + "async_get_file_checksum", +] diff --git a/src/potato_utils/io/_sync.py b/src/potato_utils/io/_sync.py new file mode 100644 index 0000000..f09597f --- /dev/null +++ b/src/potato_utils/io/_sync.py @@ -0,0 +1,264 @@ +import os +import errno +import shutil +import hashlib +import logging + +from pydantic import validate_call + +from ..constants import WarnEnum, HashAlgoEnum, MAX_PATH_LENGTH + + +logger = logging.getLogger(__name__) + + +@validate_call +def create_dir(create_dir: str, warn_mode: WarnEnum = WarnEnum.DEBUG) -> None: + """Create directory if `create_dir` doesn't exist. + + Args: + create_dir (str, required): Create directory path. + warn_mode (str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `create_dir` argument length is out of range. + OSError : When warning mode is set to ERROR and directory already exists. + OSError : If failed to create directory. + """ + + create_dir = create_dir.strip() + if (len(create_dir) < 1) or (MAX_PATH_LENGTH < len(create_dir)): + raise ValueError( + f"`create_dir` argument length {len(create_dir)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if not os.path.isdir(create_dir): + try: + _message = f"Creating '{create_dir}' directory..." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + os.makedirs(create_dir) + except OSError as err: + if (err.errno == errno.EEXIST) and (warn_mode == WarnEnum.DEBUG): + logger.debug(f"'{create_dir}' directory already exists!") + else: + logger.error(f"Failed to create '{create_dir}' directory!") + raise + + _message = f"Successfully created '{create_dir}' directory." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.EEXIST, f"'{create_dir}' directory already exists!") + + return + + +@validate_call +def remove_dir(remove_dir: str, warn_mode: WarnEnum = WarnEnum.DEBUG) -> None: + """Remove directory if `remove_dir` exists. + + Args: + remove_dir (str, required): Remove directory path. + warn_mode (str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `remove_dir` argument length is out of range. + OSError : When warning mode is set to ERROR and directory doesn't exist. + OSError : If failed to remove directory. + """ + + remove_dir = remove_dir.strip() + if (len(remove_dir) < 1) or (MAX_PATH_LENGTH < len(remove_dir)): + raise ValueError( + f"`remove_dir` argument length {len(remove_dir)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if os.path.isdir(remove_dir): + try: + _message = f"Removing '{remove_dir}' directory..." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + shutil.rmtree(remove_dir) + except OSError as err: + if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG): + logger.debug(f"'{remove_dir}' directory doesn't exist!") + else: + logger.error(f"Failed to remove '{remove_dir}' directory!") + raise + + _message = f"Successfully removed '{remove_dir}' directory." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.ENOENT, f"'{remove_dir}' directory doesn't exist!") + + return + + +@validate_call +def remove_dirs(remove_dirs: list[str], warn_mode: WarnEnum = WarnEnum.DEBUG) -> None: + """Remove directories if `remove_dirs` exist. + + Args: + remove_dirs (List[str], required): Remove directory paths as list. + warn_mode (str , optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + """ + + for _remove_dir in remove_dirs: + remove_dir(remove_dir=_remove_dir, warn_mode=warn_mode) + + return + + +@validate_call +def remove_file(file_path: str, warn_mode: WarnEnum = WarnEnum.DEBUG) -> None: + """Remove file if `file_path` exists. + + Args: + file_path (str, required): Remove file path. + warn_mode (str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `file_path` argument length is out of range. + OSError : When warning mode is set to ERROR and file doesn't exist. + OSError : If failed to remove file. + """ + + file_path = file_path.strip() + if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)): + raise ValueError( + f"`file_path` argument length {len(file_path)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if os.path.isfile(file_path): + try: + _message = f"Removing '{file_path}' file..." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + os.remove(file_path) + except OSError as err: + if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG): + logger.debug(f"'{file_path}' file doesn't exist!") + else: + logger.error(f"Failed to remove '{file_path}' file!") + raise + + _message = f"Successfully removed '{file_path}' file." + if warn_mode == WarnEnum.ALWAYS: + logger.info(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.ENOENT, f"'{file_path}' file doesn't exist!") + + return + + +@validate_call +def remove_files(file_paths: list[str], warn_mode: WarnEnum = WarnEnum.DEBUG) -> None: + """Remove files if `file_paths` exist. + + Args: + file_paths (List[str], required): Remove file paths as list. + warn_mode (str , optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + """ + + for _file_path in file_paths: + remove_file(file_path=_file_path, warn_mode=warn_mode) + + return + + +@validate_call +def get_file_checksum( + file_path: str, + hash_method: HashAlgoEnum = HashAlgoEnum.md5, + chunk_size: int = 4096, + warn_mode: WarnEnum = WarnEnum.DEBUG, +) -> str | None: + """Get file checksum. + + Args: + file_path (str , required): Target file path. + hash_method (HashAlgoEnum, optional): Hash method. Defaults to `HashAlgoEnum.md5`. + chunk_size (int , optional): Chunk size. Defaults to 4096. + warn_mode (str , optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. + Defaults to 'DEBUG'. + + Raises: + ValueError: If `file_path` argument length is out of range. + ValueError: If `chunk_size` argument value is invalid. + OSError : When warning mode is set to ERROR and file doesn't exist. + + Returns: + str | None: File checksum or None if file doesn't exist. + """ + + file_path = file_path.strip() + if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)): + raise ValueError( + f"`file_path` argument length {len(file_path)} is out of range, " + f"must be between 1 and {MAX_PATH_LENGTH} characters!" + ) + + if chunk_size < 10: + raise ValueError( + f"`chunk_size` argument value {chunk_size} is invalid, must be greater than 10!" + ) + + _file_checksum: str | None = None + if os.path.isfile(file_path): + _file_hash = hashlib.new(hash_method.value) + with open(file_path, "rb") as _file: + while True: + _file_chunk = _file.read(chunk_size) + if not _file_chunk: + break + _file_hash.update(_file_chunk) + + _file_checksum = _file_hash.hexdigest() + else: + _message = f"'{file_path}' file doesn't exist!" + if warn_mode == WarnEnum.ALWAYS: + logger.warning(_message) + elif warn_mode == WarnEnum.DEBUG: + logger.debug(_message) + elif warn_mode == WarnEnum.ERROR: + raise OSError(errno.ENOENT, _message) + + return _file_checksum + + +__all__ = [ + "create_dir", + "remove_dir", + "remove_dirs", + "remove_file", + "remove_files", + "get_file_checksum", +] diff --git a/src/potato_utils/sanitizer.py b/src/potato_utils/sanitizer.py new file mode 100644 index 0000000..f8cf059 --- /dev/null +++ b/src/potato_utils/sanitizer.py @@ -0,0 +1,85 @@ +import re +import html +from urllib.parse import quote + +from pydantic import validate_call, AnyHttpUrl + +from .constants import ( + SPECIAL_CHARS_BASE_REGEX, + SPECIAL_CHARS_LOW_REGEX, + SPECIAL_CHARS_MEDIUM_REGEX, + SPECIAL_CHARS_HIGH_REGEX, + SPECIAL_CHARS_STRICT_REGEX, +) + + +@validate_call +def escape_html(val: str) -> str: + """Escape HTML characters. + + Args: + val (str, required): String to escape. + + Returns: + str: Escaped string. + """ + + val = val.strip() + _escaped = html.escape(val) + return _escaped + + +@validate_call +def escape_url(val: AnyHttpUrl) -> str: + """Escape URL characters. + + Args: + val (AnyHttpUrl, required): String to escape. + + Returns: + str: Escaped string. + """ + + _escaped = quote(str(val)) + return _escaped + + +@validate_call +def sanitize_special_chars(val: str, mode: str = "LOW") -> str: + """Sanitize special characters. + + Args: + val (str, required): String to sanitize. + mode (str, optional): Sanitization mode. Defaults to "LOW". + + Raises: + ValueError: If `mode` is unsupported. + + Returns: + str: Sanitized string. + """ + + _pattern = r"" + mode = mode.upper().strip() + if (mode == "BASE") or (mode == "HTML"): + _pattern = SPECIAL_CHARS_BASE_REGEX + elif mode == "LOW": + _pattern = SPECIAL_CHARS_LOW_REGEX + elif mode == "MEDIUM": + _pattern = SPECIAL_CHARS_MEDIUM_REGEX + elif (mode == "HIGH") or (mode == "SCRIPT") or (mode == "SQL"): + _pattern = SPECIAL_CHARS_HIGH_REGEX + elif mode == "STRICT": + _pattern = SPECIAL_CHARS_STRICT_REGEX + else: + raise ValueError(f"Unsupported mode: {mode}") + + _sanitized = re.sub(pattern=_pattern, repl="", string=val) + return _sanitized + + +__all__ = [ + "escape_html", + "escape_url", + "sanitize_special_chars", +] diff --git a/src/potato_utils/secure.py b/src/potato_utils/secure.py new file mode 100644 index 0000000..1a45310 --- /dev/null +++ b/src/potato_utils/secure.py @@ -0,0 +1,90 @@ +import uuid +import string +import secrets +import hashlib + +from pydantic import validate_call + +from .constants import HashAlgoEnum +from .dt import now_ts + + +@validate_call +def gen_unique_id(prefix: str = "") -> str: + """Generate unique id. Format: '{prefix}{datetime}_{uuid4}'. + + Args: + prefix (str, optional): Prefix of id. Defaults to ''. + + Raises: + ValueError: If `prefix` length is greater than 32. + + Returns: + str: Unique id. + """ + + prefix = prefix.strip() + if 32 < len(prefix): + raise ValueError( + f"`prefix` argument length {len(prefix)} is too long, must be less than or equal to 32!", + ) + + _id = str(f"{prefix}{now_ts()}_{uuid.uuid4().hex}").lower() + return _id + + +@validate_call +def gen_random_string(length: int = 16, is_alphanum: bool = True) -> str: + """Generate secure random string. + + Args: + length (int , optional): Length of random string. Defaults to 16. + is_alphanum (bool, optional): If True, generate only alphanumeric string. Defaults to True. + + Raises: + ValueError: If `length` is less than 1. + + Returns: + str: Generated random string. + """ + + if length < 1: + raise ValueError( + f"`length` argument value {length} is too small, must be greater than or equal to 1!", + ) + + _base_chars = string.ascii_letters + string.digits + if not is_alphanum: + _base_chars += string.punctuation + + _random_str = "".join(secrets.choice(_base_chars) for _i in range(length)) + return _random_str + + +@validate_call +def hash_str(val: str | bytes, algorithm: HashAlgoEnum = HashAlgoEnum.sha256) -> str: + """Hash a string using a specified hash algorithm. + + Args: + val (str | bytes , required): The value to be hashed. + algorithm (HashAlgoEnum, required): The hash algorithm to use. Defaults to `HashAlgoEnum.sha256`. + + Returns: + str: The hexadecimal representation of the digest. + """ + + if isinstance(val, str): + val = val.encode("utf-8") + + _hash = hashlib.new(algorithm.value) + _hash.update(val) + + _hash_val = _hash.hexdigest() + return _hash_val + + +__all__ = [ + "gen_unique_id", + "gen_random_string", + "hash_str", +] diff --git a/src/potato_utils/validator.py b/src/potato_utils/validator.py new file mode 100644 index 0000000..e0e81ac --- /dev/null +++ b/src/potato_utils/validator.py @@ -0,0 +1,150 @@ +import re +from re import Pattern + +from pydantic import validate_call + +from .constants import ( + REQUEST_ID_REGEX, + SPECIAL_CHARS_BASE_REGEX, + SPECIAL_CHARS_LOW_REGEX, + SPECIAL_CHARS_MEDIUM_REGEX, + SPECIAL_CHARS_HIGH_REGEX, + SPECIAL_CHARS_STRICT_REGEX, +) + + +@validate_call +def is_truthy(val: str | bool | int | float | None) -> bool: + """Check if the value is truthy. + + Args: + val (Union[str, bool, int, float, None], required): Value to check. + + Raises: + ValueError: If `val` argument type is string and value is invalid. + + Returns: + bool: True if the value is truthy, False otherwise. + """ + + if isinstance(val, str): + val = val.strip().lower() + + if val in ["0", "false", "f", "no", "n", "off"]: + return False + elif val in ["1", "true", "t", "yes", "y", "on"]: + return True + else: + raise ValueError(f"`val` argument value is invalid: '{val}'!") + + return bool(val) + + +@validate_call +def is_falsy(val: str | bool | int | float | None) -> bool: + """Check if the value is falsy. + + Args: + val (Union[str, bool, int, float, None], required): Value to check. + + Returns: + bool: True if the value is falsy, False otherwise. + """ + + return not is_truthy(val) + + +@validate_call +def is_request_id(val: str) -> bool: + """Check if the string is valid request ID. + + Args: + val (str, required): String to check. + + Returns: + bool: True if the string is valid request ID, False otherwise. + """ + + _is_valid = bool(re.match(pattern=REQUEST_ID_REGEX, string=val)) + return _is_valid + + +@validate_call +def is_blacklisted(val: str, blacklist: list[str]) -> bool: + """Check if the string is blacklisted. + + Args: + val (str , required): String to check. + blacklist (List[str], required): List of blacklisted strings. + + Returns: + bool: True if the string is blacklisted, False otherwise. + """ + + for _blacklisted in blacklist: + if _blacklisted in val: + return True + + return False + + +@validate_call +def is_valid(val: str, pattern: Pattern | str) -> bool: + """Check if the string is valid with given pattern. + + Args: + val (str , required): String to check. + pattern (Union[Pattern, str], required): Pattern regex to check. + + Returns: + bool: True if the string is valid with given pattern, False otherwise. + """ + + _is_valid = bool(re.match(pattern=pattern, string=val)) + return _is_valid + + +@validate_call +def has_special_chars(val: str, mode: str = "LOW") -> bool: + """Check if the string has special characters. + + Args: + val (str, required): String to check. + mode (str, optional): Check mode. Defaults to "LOW". + + Raises: + ValueError: If `mode` is unsupported. + + Returns: + bool: True if the string has special characters, False otherwise. + """ + + _has_special_chars = False + + _pattern = r"" + mode = mode.upper().strip() + if (mode == "BASE") or (mode == "HTML"): + _pattern = SPECIAL_CHARS_BASE_REGEX + elif mode == "LOW": + _pattern = SPECIAL_CHARS_LOW_REGEX + elif mode == "MEDIUM": + _pattern = SPECIAL_CHARS_MEDIUM_REGEX + elif (mode == "HIGH") or (mode == "SCRIPT") or (mode == "SQL"): + _pattern = SPECIAL_CHARS_HIGH_REGEX + elif mode == "STRICT": + _pattern = SPECIAL_CHARS_STRICT_REGEX + else: + raise ValueError(f"Unsupported mode: {mode}") + + _has_special_chars = bool(re.search(pattern=_pattern, string=val)) + return _has_special_chars + + +__all__ = [ + "is_truthy", + "is_falsy", + "is_request_id", + "is_blacklisted", + "is_valid", + "has_special_chars", +] diff --git a/templates/configs/config.json b/templates/configs/config.json deleted file mode 100644 index 1af11ab..0000000 --- a/templates/configs/config.json +++ /dev/null @@ -1,9 +0,0 @@ -{ - "potato_utils": { - "min_length": 2, - "max_length": 100, - "min_value": 0.0, - "max_value": 1.0, - "threshold": 0.5 - } -} diff --git a/templates/configs/config.yml b/templates/configs/config.yml deleted file mode 100644 index 0396acc..0000000 --- a/templates/configs/config.yml +++ /dev/null @@ -1,6 +0,0 @@ -potato_utils: - min_length: 2 - max_length: 100 - min_value: 0.0 - max_value: 1.0 - threshold: 0.5 diff --git a/tests/test_potato_utils.py b/tests/test_potato_utils.py index be44339..a00d076 100644 --- a/tests/test_potato_utils.py +++ b/tests/test_potato_utils.py @@ -1,28 +1,17 @@ import logging -import pytest +# import pytest -try: - from potato_utils import MyClass -except ImportError: - from src.potato_utils import MyClass +# try: +# from potato_utils import io as io_utils +# except ImportError: +# from potato_utils import io as io_utils logger = logging.getLogger(__name__) -@pytest.fixture -def my_object(): - _my_object = MyClass() +def test_init(): + logger.info("Testing initialization of 'potato_utils'...") - yield _my_object - - del _my_object - - -def test_init(my_object): - logger.info("Testing initialization of 'MyClass'...") - - assert isinstance(my_object, MyClass) - - logger.info("Done: Initialization of 'MyClass'.\n") + logger.info("Done: Initialization of 'potato_utils'.\n")