jarvis/jarvis-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add HTTP utils, hash utility, bump to 0.2.0

Browse source
This commit is contained in:
Zeva Rose 2022-02-09 13:49:15 -07:00
parent c5a2afc68f
commit 6f138e36fe
5 changed files with 139 additions and 141 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

130
jarvis_core/util.py
View file

@ -1,130 +0,0 @@
"""JARVIS quality of life utilities."""
from typing import Any, Callable, Iterable, List, Optional
class Singleton(object):
REQUIRED = []
OPTIONAL = {}
def __new__(cls, *args: list, **kwargs: dict):
"""Create a new singleton."""
inst = cls.__dict__.get("inst")
if inst is not None:
return inst
inst = object.__new__(cls)
inst.init(*args, **kwargs)
inst._validate()
cls.__inst__ = inst
return inst
def _validate(self) -> None:
for key in self.REQUIRED:
if not getattr(self, key, None):
raise ValueError(f"Missing required key: {key}")
def init(self, **kwargs: dict) -> None:
"""Initialize the object."""
for key, value in kwargs.items():
setattr(self, key, value)
for key, value in self.OPTIONAL.items():
if not getattr(self, key, None):
setattr(self, key, value)
def find(predicate: Callable, sequence: Iterable) -> Optional[Any]:
"""
Find the first element in a sequence that matches the predicate.
??? Hint "Example Usage:"
```python
member = find(lambda m: m.name == "UserName", guild.members)
```
Args:
predicate: A callable that returns a boolean value
sequence: A sequence to be searched
Returns:
A match if found, otherwise None
"""
for el in sequence:
if predicate(el):
return el
return None
def find_all(predicate: Callable, sequence: Iterable) -> List[Any]:
"""
Find all elements in a sequence that match the predicate.
??? Hint "Example Usage:"
```python
members = find_all(lambda m: m.name == "UserName", guild.members)
```
Args:
predicate: A callable that returns a boolean value
sequence: A sequence to be searched
Returns:
A list of matches
"""
return [el for el in sequence if predicate(el)]
def get(sequence: Iterable, **kwargs: Any) -> Optional[Any]:
"""
Find the first element in a sequence that matches all attrs.
??? Hint "Example Usage:"
```python
channel = get(guild.channels, nsfw=False, category="General")
```
Args:
sequence: A sequence to be searched
kwargs: Keyword arguments to search the sequence for
Returns:
A match if found, otherwise None
"""
if not kwargs:
return sequence[0]
for el in sequence:
if any(not hasattr(el, attr) for attr in kwargs.keys()):
continue
if all(getattr(el, attr) == value for attr, value in kwargs.items()):
return el
return None
def get_all(sequence: Iterable, **kwargs: Any) -> List[Any]:
"""
Find all elements in a sequence that match all attrs.
??? Hint "Example Usage:"
```python
channels = get_all(guild.channels, nsfw=False, category="General")
```
Args:
sequence: A sequence to be searched
kwargs: Keyword arguments to search the sequence for
Returns:
A list of matches
"""
if not kwargs:
return sequence
matches = []
for el in sequence:
if any(not hasattr(el, attr) for attr in kwargs.keys()):
continue
if all(getattr(el, attr) == value for attr, value in kwargs.items()):
matches.append(el)
return matches

101
jarvis_core/util/__init__.py Normal file
View file

@ -0,0 +1,101 @@
"""JARVIS quality of life utilities."""
import hashlib
from typing import Callable, Tuple, Union
from aiohttp import ClientSession
from jarvis_core.filters import url
DEFAULT_BLOCKSIZE = 8 * 1024 * 1024
class Singleton(object):
REQUIRED = []
OPTIONAL = {}
def __new__(cls, *args: list, **kwargs: dict):
"""Create a new singleton."""
inst = cls.__dict__.get("inst")
if inst is not None:
return inst
inst = object.__new__(cls)
inst.init(*args, **kwargs)
inst._validate()
cls.__inst__ = inst
return inst
def _validate(self) -> None:
for key in self.REQUIRED:
if not getattr(self, key, None):
raise ValueError(f"Missing required key: {key}")
def init(self, **kwargs: dict) -> None:
"""Initialize the object."""
for key, value in kwargs.items():
setattr(self, key, value)
for key, value in self.OPTIONAL.items():
if not getattr(self, key, None):
setattr(self, key, value)
async def hash(
data: str, method: Union[Callable, str] = hashlib.sha256, size: int = DEFAULT_BLOCKSIZE
) -> Tuple[str, int, str]:
"""
Hash an object.
Args:
hash_func: Hash function, default `hashlib.sha256`
data: URL or text to hash
Returns:
Tuple of hash hex, data size, and content type
Raises:
Exception: Raise on status
"""
# Get hashlib method
if isinstance(method, str):
method = getattr(hashlib, method, hashlib.sha256)
method = method()
# Hash non-URLs
if not url.match(data):
method.update(data.encode("UTF8"))
return method.hexdigest(), len(data), "string"
# Hash URL contents
async with ClientSession() as session:
resp = await session.get(data)
resp.raise_for_status()
content_type = resp.content_type
data_len = resp.content_length
async for block in resp.content.iter_chunked(n=size):
method.update(block)
return method.hexdigest(), data_len, content_type
def convert_bytesize(b: int) -> str:
"""Convert bytes amount to human readable."""
b = float(b)
sizes = ["B", "KB", "MB", "GB", "TB", "PB"]
size = 0
while b >= 1024 and size < len(sizes) - 1:
b = b / 1024
size += 1
return "{:0.3f} {}".format(b, sizes[size])
def unconvert_bytesize(size: int, ending: str) -> int:
"""Convert human readable to bytes."""
ending = ending.upper()
sizes = ["B", "KB", "MB", "GB", "TB", "PB"]
if ending == "B":
return size
# Rounding is only because bytes cannot be partial
return round(size * (1024 ** sizes.index(ending)))

27
jarvis_core/util/http.py Normal file
View file

@ -0,0 +1,27 @@
"""HTTP helper methods."""
from aiohttp import ClientSession
from jarvis_core.filters import url
async def get_size(link: str) -> int:
"""
Get the url filesize.
Args:
link: URL to get
Returns:
Size in bytes
Raises:
Exception: On status code
ValueError: On bad URL
"""
if not url.match(link):
raise ValueError("Invalid URL.")
async with ClientSession() as session:
resp = await session.get(link)
resp.raise_for_status()
return resp.content_length

20
poetry.lock generated
View file

@ -104,7 +104,7 @@ python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
[[package]]
name = "dis-snek"
version = "5.0.0"
version = "6.0.0"
description = "An API wrapper for Discord filled with snakes"
category = "main"
optional = false
@ -266,7 +266,7 @@ testing = ["docopt", "pytest (<6.0.0)"]
[[package]]
name = "platformdirs"
version = "2.4.1"
version = "2.5.0"
description = "A small Python module for determining appropriate platform-specific dirs, e.g. a \"user data dir\"."
category = "dev"
optional = false
@ -479,7 +479,7 @@ python-versions = ">=2.6, !=3.0.*, !=3.1.*, !=3.2.*"
[[package]]
name = "tomli"
version = "2.0.0"
version = "2.0.1"
description = "A lil' TOML parser"
category = "main"
optional = false
@ -549,7 +549,7 @@ multidict = ">=4.0"
[metadata]
lock-version = "1.1"
python-versions = "^3.10"
content-hash = "f3033dbd5b68c51984d71fe994f3ac8ea040a65e8fa7d174c209c38be5450973"
content-hash = "8edf07b473e9615a836f71510d792021c6760ef94abe4b720f9ff12a37c5dce5"
[metadata.files]
aiohttp = [
@ -659,8 +659,8 @@ colorama = [
{file = "colorama-0.4.4.tar.gz", hash = "sha256:5941b2b48a20143d2267e95b1c2a7603ce057ee39fd88e7329b0c292aa16869b"},
]
dis-snek = [
{file = "dis-snek-5.0.0.tar.gz", hash = "sha256:cc733b510d6b20523a8067f19b6d9b99804c13e37d78cd6e0fc401098adbca27"},
{file = "dis_snek-5.0.0-py3-none-any.whl", hash = "sha256:d1d50ba468ad6b0788e9281eb9d83f6eb2f8d964c1212ccd0e3fb33295462263"},
{file = "dis-snek-6.0.0.tar.gz", hash = "sha256:3abe2af832bd87adced01ebc697e418b25871a4158ac8ba2e202d8fbb2921f44"},
{file = "dis_snek-6.0.0-py3-none-any.whl", hash = "sha256:106cb08b0cc982c59db31be01ee529e4d7273835de7f418562d8e6b24d7f965d"},
]
flake8 = [
{file = "flake8-4.0.1-py2.py3-none-any.whl", hash = "sha256:479b1304f72536a55948cb40a32dce8bb0ffe3501e26eaf292c7e60eb5e0428d"},
@ -890,8 +890,8 @@ parso = [
{file = "parso-0.8.3.tar.gz", hash = "sha256:8c07be290bb59f03588915921e29e8a50002acaf2cdc5fa0e0114f91709fafa0"},
]
platformdirs = [
{file = "platformdirs-2.4.1-py3-none-any.whl", hash = "sha256:1d7385c7db91728b83efd0ca99a5afb296cab9d0ed8313a45ed8ba17967ecfca"},
{file = "platformdirs-2.4.1.tar.gz", hash = "sha256:440633ddfebcc36264232365d7840a970e75e1018d15b4327d11f91909045fda"},
{file = "platformdirs-2.5.0-py3-none-any.whl", hash = "sha256:30671902352e97b1eafd74ade8e4a694782bd3471685e78c32d0fdfd3aa7e7bb"},
{file = "platformdirs-2.5.0.tar.gz", hash = "sha256:8ec11dfba28ecc0715eb5fb0147a87b1bf325f349f3da9aab2cd6b50b96b692b"},
]
pluggy = [
{file = "pluggy-0.13.1-py2.py3-none-any.whl", hash = "sha256:966c145cd83c96502c3c3868f50408687b38434af77734af1e9ca461a4081d2d"},
@ -1090,8 +1090,8 @@ toml = [
{file = "toml-0.10.2.tar.gz", hash = "sha256:b3bda1d108d5dd99f4a20d24d9c348e91c4db7ab1b749200bded2f839ccbe68f"},
]
tomli = [
{file = "tomli-2.0.0-py3-none-any.whl", hash = "sha256:b5bde28da1fed24b9bd1d4d2b8cba62300bfb4ec9a6187a957e8ddb9434c5224"},
{file = "tomli-2.0.0.tar.gz", hash = "sha256:c292c34f58502a1eb2bbb9f5bbc9a5ebc37bee10ffb8c2d6bbdfa8eb13cc14e1"},
{file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"},
{file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"},
]
ujson = [
{file = "ujson-5.1.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:644552d1e89983c08d0c24358fbcb5829ae5b5deee9d876e16d20085cfa7dc81"},

2
pyproject.toml
View file

@ -1,6 +1,6 @@
[tool.poetry]
name = "jarvis-core"
version = "0.1.7"
version = "0.2.0"
description = ""
authors = ["Your Name <you@example.com>"]