add basic utils

.github/workflows/release.yml

@@ -32,7 +32,6 @@ jobs:
           pip install -r requirements_build.txt >/dev/null
           pip install -r requirements.txt >/dev/null
 
-      # NOTE: timeout for running the app includes db migrations
       - name: Testing to build with PIP
         run: |
           cd /tmp

.gitignore

@@ -1,2 +1,4 @@
 __pychache__
-venv/
+venv/
+.idea/
+src/oxl_utils.egg-info

README.md

@@ -1,2 +1,39 @@
 # Utils Collection
 
+## Data States
+
+```python3
+from oxl_utils.state import is_set
+from oxl_utils.state import is_null
+```
+
+## Validators
+
+```python3
+# validate email format
+from oxl_utils.valid.email import valid_email
+
+# also check for valid MX record of e-mail domain
+from oxl_utils.valid.email import valid_email_dns
+
+# ips and networks
+from oxl_utils.valid.net import valid_ip
+from oxl_utils.valid.net import valid_ip4
+from oxl_utils.valid.net import valid_ip6
+from oxl_utils.valid.net import valid_net4
+from oxl_utils.valid.net import valid_net6
+from oxl_utils.valid.net import valid_public_ip
+from oxl_utils.valid.net import valid_asn
+
+from oxl_utils.valid.dns import valid_domain
+from oxl_utils.valid.email import valid_email
+from oxl_utils.valid.email import has_mailserver
+
+```
+
+## Django
+
+```python3
+# fix datetime timezone
+from oxl_utils.dj.dt import datetime_from_db
+```

requirements.txt

@@ -1 +1,2 @@
-pytz
+pytz
+dnspython

src/oxl_utils/debug.py

@@ -0,0 +1,5 @@
+from os import environ
+
+
+def mode_debug() -> bool:
+    return 'DEBUG' in environ and environ['DEBUG'] == '1'

src/oxl_utils/dj/dt.py

@@ -0,0 +1,23 @@
+from os import environ
+from datetime import datetime
+
+from pytz import utc, timezone
+
+
+def datetime_from_db(dt: (datetime, None), tz: (timezone, str) = None) -> (datetime, None):
+    if not isinstance(dt, datetime):
+        return None
+
+    if tz in [None, '']:
+        try:
+            tz = environ['TIMEZONE']
+
+        except KeyError:
+            raise EnvironmentError('TIMEZONE not provided')
+
+    if isinstance(tz, str):
+        tz = timezone(tz)
+
+    # datetime form db will always be UTC; convert it
+    local_dt = dt.replace(tzinfo=utc).astimezone(tz)
+    return tz.normalize(local_dt)

src/oxl_utils/dj/model.py

@@ -0,0 +1,5 @@
+CHOICES_BOOL = (
+    (True, 'Yes'),
+    (False, 'No')
+)
+DEFAULT_NONE = {'null': True, 'default': None, 'blank': True}

src/oxl_utils/dt.py

@@ -0,0 +1,18 @@
+from os import environ
+from datetime import datetime
+
+from pytz import timezone
+
+
+def datetime_w_tz(tz: (timezone, str) = None) -> datetime:
+    if tz in [None, '']:
+        try:
+            tz = environ['TIMEZONE']
+
+        except KeyError:
+            raise EnvironmentError('TIMEZONE not provided')
+
+    if isinstance(tz, str):
+        tz = timezone(tz)
+
+    return datetime.now(tz)

src/oxl_utils/log.py

@@ -0,0 +1,56 @@
+from os import getpid, environ
+from sys import stderr, stdout
+from inspect import stack as inspect_stack
+from inspect import getfile as inspect_getfile
+
+from .dt import datetime_w_tz
+from .debug import mode_debug
+
+LOG_FORMAT = 'default' if 'LOG_FORMAT' not in environ else environ['LOG_FORMAT']
+LOG_TIME_FORMAT = '%Y-%m-%d %H:%M:%S %z'
+PID = getpid()
+
+LEVEL_NAME_MAPPING = {
+    1: 'FATAL',
+    2: 'ERROR',
+    3: 'WARN',
+    4: 'INFO',
+    5: 'INFO',
+    6: 'DEBUG',
+    7: 'DEBUG',
+}
+
+
+def _log_formatted(level: int, msg: str, mid: str = '') -> str:
+    if LOG_FORMAT == 'gunicorn':
+        return f"[{datetime_w_tz().strftime(LOG_TIME_FORMAT)}] [{PID}] [{LEVEL_NAME_MAPPING[level]}] {mid}{msg}"
+
+    return f"{datetime_w_tz().strftime(LOG_TIME_FORMAT)} {LEVEL_NAME_MAPPING[level]} {mid}{msg}"
+
+
+def log(msg: str, level: int = 3):
+    debug = mode_debug()
+    prefix_caller = ''
+
+    if level > 5 and not debug:
+        return
+
+    if debug:
+        caller = inspect_getfile(inspect_stack()[1][0]).rsplit('/', 1)[1].rsplit('.', 1)[0]
+        prefix_caller = f'[{caller}] '
+
+    stdout.write(_log_formatted(msg=msg, level=level, mid=prefix_caller))
+
+
+def log_warn(msg: str, _stderr: bool = False):
+    msg = f'\x1b[1;33m{_log_formatted(msg=msg, level=3)}\x1b[0m\n'
+
+    if _stderr:
+        stderr.write(msg)
+
+    else:
+        stdout.write(msg)
+
+
+def log_error(msg: str):
+    stderr.write(f'\033[01;{_log_formatted(msg=msg, level=2)}\x1b[0m\n')

src/oxl_utils/net.py

@@ -0,0 +1,57 @@
+from socket import socket, AF_INET, AF_INET6, SOCK_STREAM
+from os import environ
+
+from dns.resolver import Resolver, NoAnswer, NXDOMAIN, LifetimeTimeout, NoNameservers
+from dns.exception import SyntaxError as DNSSyntaxError
+
+from .valid.net import valid_ip
+
+DEFAULT_NAMESERVERS = ['1.1.1.1', '8.8.8.8']
+NS_ENV_KEY = 'NAMESERVERS'
+
+dns_resolver = Resolver(configure=False)
+dns_resolver.lifetime = 0.1
+dns_resolver.timeout = 0.1
+
+dns_resolver.nameservers = environ[NS_ENV_KEY].split(',') if NS_ENV_KEY in environ else DEFAULT_NAMESERVERS
+
+
+def resolve_dns(v: str, t: str = 'A', timeout: float = dns_resolver.timeout) -> list[str]:
+    try:
+        if t != 'PTR':
+            r = [r.to_text() for r in dns_resolver.resolve(v, t, lifetime=timeout)]
+
+        else:
+            r = [r.to_text() for r in dns_resolver.resolve_address(v, lifetime=timeout)]
+
+        r.sort()
+        return r
+
+    except (IndexError, NoAnswer, NXDOMAIN, DNSSyntaxError, NoNameservers, LifetimeTimeout):
+        return []
+
+
+def resolve_first_ip(v: str) -> (str, None):
+    r = resolve_dns(v, t='A')
+    if len(r) > 0:
+        return r[0]
+
+    r = resolve_dns(v, t='AAAA')
+    if len(r) > 0:
+        return r[0]
+
+    return None
+
+
+def is_port_open(target: str, port: (str, int), timeout: int = 1) -> bool:
+    ip = target
+    if not valid_ip(target):
+        ip = resolve_first_ip(target)
+        if ip is None:
+            return False
+
+    ip_proto = AF_INET if ip.find(':') == -1 else AF_INET6
+
+    with socket(ip_proto, SOCK_STREAM) as s:
+        s.settimeout(timeout)
+        return s.connect_ex((ip, int(port))) == 0

src/oxl_utils/net_test.py

@@ -0,0 +1,64 @@
+import pytest
+
+
+@pytest.mark.parametrize('v, s', [
+    (
+            {'v': 'oxl.at', 't': 'A'},
+            {'exist': True, 'len': 1},
+    ),
+    (
+            {'v': 'oxl.at', 't': 'AAAA'},
+            {'exist': True, 'len': 1},
+    ),
+    (
+            {'v': 'xyz.oxl.at', 't': 'MX'},
+            {'exist': False, 'len': 0},
+    ),
+    (
+            {'v': 'xyz.oxl.at', 't': 'MX'},
+            {'exist': False, 'len': 0},
+    ),
+    (
+            {'v': '1.1.1.1', 't': 'PTR'},
+            {'exist': True, 'len': 1, 'v': ['one.one.one.one.']},
+    ),
+    (
+            {'v': 'one.one.one.one', 't': 'A'},
+            {'exist': True, 'len': 2, 'v': ['1.0.0.1', '1.1.1.1']},
+    ),
+])
+def test_dns(v: dict, s: dict):
+    from .net import resolve_dns
+    r = resolve_dns(**v, timeout=2.0)
+    assert (len(r) > 0) == s['exist']
+    assert len(r) == s['len']
+    if 'v' in s:
+        assert r == s['v']
+
+
+@pytest.mark.parametrize('v, s', [
+    ('one.one.one.one', '1.0.0.1'),
+    ('test.abc.sldjflsdkl.sdlfj', None),
+])
+def test_dns_first_ip(v: str, s: str):
+    from .net import resolve_first_ip
+    assert resolve_first_ip(v) == s
+
+
+@pytest.mark.parametrize('v, s', [
+    (
+            {'target': '1.1.1.1', 'port': 53},
+            True,
+    ),
+    (
+            {'target': '1.1.1.1', 'port': 54},
+            False,
+    ),
+    (
+            {'target': 'one.one.one.one', 'port': 53},
+            True,
+    ),
+])
+def test_port(v: dict, s: bool):
+    from .net import is_port_open
+    assert is_port_open(**v) == s

src/oxl_utils/state.py

@@ -0,0 +1,9 @@
+def is_null(data) -> bool:
+    if data is None:
+        return True
+
+    return str(data).strip() == ''
+
+
+def is_set(data: str) -> bool:
+    return not is_null(data)

src/oxl_utils/state_test.py

@@ -0,0 +1,31 @@
+import pytest
+
+
+@pytest.mark.parametrize('v, r', [
+    ('', True),
+    (' ', True),
+    ('                ', True),
+    (None, True),
+    ('test', False),
+    (1, False),
+    (True, False),
+    ('None', False),
+])
+def test_null(v: any, r: bool):
+    from .state import is_null
+    assert is_null(v) == r
+
+
+@pytest.mark.parametrize('v, r', [
+    ('', False),
+    (' ', False),
+    ('                ', False),
+    (None, False),
+    ('test', True),
+    (1, True),
+    (True, True),
+    ('None', True),
+])
+def test_set(v: any, r: bool):
+    from .state import is_set
+    assert is_set(v) == r