Compare commits

..

10 commits

20 changed files with 479 additions and 59 deletions

1
.gitignore vendored
View file

@ -30,6 +30,7 @@ docs/_build
docs/_static docs/_static
docs/templates docs/templates
.coverage .coverage
.pytest_cache/
# changes during CD/CI # changes during CD/CI
aliases/*/pyproject.toml aliases/*/pyproject.toml

View file

@ -1,5 +1,26 @@
# Changelog # Changelog
## 0.14.0
* Added `ast` module
* Deprecate `dei_args()` for problems with the typing system. The function is not going away tho
* Module `sqlalchemy`: added `email_column()`
## 0.13.1 and 0.12.7
+ Typing fixes
## 0.13.0 "Laconic Letters"
+ Added module `argparse` with class `LetterSubparsers()`, which allows pacman-style args by preprocessing them
before feeding them to parse_args
+ Module `sqlalchemy`:
* removed deprecated alias `entity_base()`. use `declarative_base()` instead.
* fix imports.
+ Module `functools`: add `cooldown()`, `do_not_flood()`
+ Module `color`: add `ColorFormatter()`
+ Separated `suou[waiter]` dependency from `suou[quart]`
## 0.12.6 ## 0.12.6
+ Added unittests to `dei_args()` + Added unittests to `dei_args()`

View file

@ -1,6 +1,6 @@
# SIS Unified Object Underarmor # SIS Unified Object Underarmor
Good morning, my brother! Welcome **SUOU** (**S**IS **U**nified **O**bject **U**nderarmor), the Python library which speeds up and makes it pleasing to develop API, database schemas and stuff in Python. Good morning, my brother! Welcome **SUOU** (**S**IS **U**nified **O**bject **U**nderarmor), the Python library which (maybe) speeds up and makes it pleasing to develop API, database schemas and stuff in Python.
It provides utilities such as: It provides utilities such as:
* SIQ ([specification](https://yusur.moe/protocols/siq.html) \[LINK BROKEN - WON'T FIX\] - [copy](https://suou.readthedocs.io/en/latest/iding.html)) * SIQ ([specification](https://yusur.moe/protocols/siq.html) \[LINK BROKEN - WON'T FIX\] - [copy](https://suou.readthedocs.io/en/latest/iding.html))
@ -15,13 +15,13 @@ It provides utilities such as:
**Python 3.10**+ with Pip is required. **Python 3.10**+ with Pip is required.
```bash ```bash
$ pip install sakuragasaki46-suou $ pip install suou
``` ```
To install optional dependencies (i.e. `sqlalchemy`) for development use: To install optional dependencies (i.e. `sqlalchemy`) for development use:
```bash ```bash
$ pip install sakuragasaki46-suou[sqlalchemy] $ pip install suou[sqlalchemy]
``` ```
Please note that you probably already have those dependencies, if you just use the library. Please note that you probably already have those dependencies, if you just use the library.
@ -34,7 +34,7 @@ Read the [documentation](https://suou.readthedocs.io/).
### Disclaimer ### Disclaimer
Just a heads up: SUOU was made to support Sakuragasaki46 (me)'s own selfish, egoistic needs. Not certainly to provide a service to the public. Just a heads up: SUOU was made to support yusurko (me)'s own selfish, egoistic needs. Not certainly to provide a service to the public.
As a consequence, 'add this add that' stuff is best-effort. As a consequence, 'add this add that' stuff is best-effort.
@ -42,8 +42,6 @@ Expect breaking changes, disruptive renames in bugfix releases, sudden deprecati
Don't want to depend on my codebase for moral reasons (albeit unrelated)? It's fine. I did not ask you. Don't want to depend on my codebase for moral reasons (albeit unrelated)? It's fine. I did not ask you.
**DO NOT ASK TO MAKE SUOU SAFE FOR CHILDREN**. Enjoy having your fingers cut.
### "LTS" ### "LTS"
The following versions are supported: the latest, the second-to-latest, 0.12.x and 0.7.x. The following versions are supported: the latest, the second-to-latest, 0.12.x and 0.7.x.
@ -54,7 +52,7 @@ Licensed under the [Apache License, Version 2.0](LICENSE), a non-copyleft free a
This is a hobby project, made available “AS IS”, with __no warranty__ express or implied. This is a hobby project, made available “AS IS”, with __no warranty__ express or implied.
I (sakuragasaki46) may NOT be held accountable for Your use of my code. I (yusurko) may NOT be held accountable for Your use of my code.
> It's pointless to file a lawsuit because you feel damaged, and it's only going to turn against you. What a waste of money you could have spent on a vacation or charity, or invested in stocks. > It's pointless to file a lawsuit because you feel damaged, and it's only going to turn against you. What a waste of money you could have spent on a vacation or charity, or invested in stocks.

View file

@ -56,7 +56,9 @@ markdown = [
] ]
quart = [ quart = [
"Quart", "Quart",
"Quart-Schema", "Quart-Schema"
]
waiter = [
"starlette>=0.47.2" "starlette>=0.47.2"
] ]
quart_auth = [ quart_auth = [
@ -78,7 +80,8 @@ full = [
"suou[quart_auth]", # includes quart, sqlalchemy and quart_sqlalchemy "suou[quart_auth]", # includes quart, sqlalchemy and quart_sqlalchemy
"suou[peewee]", "suou[peewee]",
"suou[markdown]", "suou[markdown]",
"suou[sass]" "suou[sass]",
"suou[waiter]"
] ]
docs = [ docs = [

View file

@ -18,13 +18,14 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
from .iding import Siq, SiqCache, SiqType, SiqGen from .iding import Siq, SiqCache, SiqType, SiqGen
from .codecs import (StringCase, cb32encode, cb32decode, b32lencode, b32ldecode, b64encode, b64decode, b2048encode, b2048decode, from .codecs import (StringCase, cb32encode, cb32decode, b32lencode, b32ldecode, b64encode, b64decode, b2048encode, b2048decode,
jsonencode, twocolon_list, want_bytes, want_str, ssv_list, want_urlsafe, want_urlsafe_bytes) jsonencode, twocolon_list, want_bytes, want_str, ssv_list, want_urlsafe, want_urlsafe_bytes,
z85encode, z85decode)
from .bits import count_ones, mask_shift, split_bits, join_bits, mod_ceil, mod_floor from .bits import count_ones, mask_shift, split_bits, join_bits, mod_ceil, mod_floor
from .calendar import want_datetime, want_isodate, want_timestamp, age_and_days from .calendar import want_datetime, want_isodate, want_timestamp, age_and_days
from .configparse import MissingConfigError, MissingConfigWarning, ConfigOptions, ConfigParserConfigSource, ConfigSource, DictConfigSource, ConfigValue, EnvConfigSource from .configparse import MissingConfigError, MissingConfigWarning, ConfigOptions, ConfigParserConfigSource, ConfigSource, DictConfigSource, ConfigValue, EnvConfigSource
from .collections import TimedDict from .collections import TimedDict
from .dei import dei_args from .dei import dei_args
from .functools import deprecated, not_implemented, timed_cache, none_pass, alru_cache, future from .functools import deprecated, not_implemented, timed_cache, none_pass, alru_cache, future, cooldown, do_not_flood
from .classtools import Wanted, Incomplete from .classtools import Wanted, Incomplete
from .itertools import makelist, kwargs_prefix, ltuple, rtuple, additem, addattr from .itertools import makelist, kwargs_prefix, ltuple, rtuple, additem, addattr
from .i18n import I18n, JsonI18n, TomlI18n from .i18n import I18n, JsonI18n, TomlI18n
@ -32,18 +33,21 @@ from .signing import UserSigner
from .snowflake import Snowflake, SnowflakeGen from .snowflake import Snowflake, SnowflakeGen
from .lex import symbol_table, lex, ilex from .lex import symbol_table, lex, ilex
from .strtools import PrefixIdentifier from .strtools import PrefixIdentifier
from .validators import matches, not_less_than, not_greater_than, yesno from .validators import matches, not_less_than, not_greater_than, yesno, must_be
from .redact import redact_url_password from .redact import redact_url_password
from .http import WantsContentType from .http import WantsContentType
from .color import OKLabColor, chalk, WebColor, RGBColor, LinearRGBColor, XYZColor, OKLCHColor from .color import OKLabColor, chalk, WebColor, RGBColor, LinearRGBColor, \
XYZColor, OKLCHColor, ColorFormatter
from .mat import Matrix from .mat import Matrix
from .argparse import LetterSubparsers
__version__ = "0.12.6" __version__ = "0.14.0a1"
__all__ = ( __all__ = (
'ColorFormatter',
'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue', 'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue',
'DictConfigSource', 'EnvConfigSource', 'I18n', 'Incomplete', 'JsonI18n', 'DictConfigSource', 'EnvConfigSource', 'I18n', 'Incomplete', 'JsonI18n',
'LinearRGBColor', 'LetterSubparsers', 'LinearRGBColor',
'Matrix', 'MissingConfigError', 'MissingConfigWarning', 'OKLabColor', 'OKLCHColor', 'Matrix', 'MissingConfigError', 'MissingConfigWarning', 'OKLabColor', 'OKLCHColor',
'PrefixIdentifier', 'RGBColor', 'PrefixIdentifier', 'RGBColor',
'Siq', 'SiqCache', 'SiqGen', 'SiqType', 'Snowflake', 'SnowflakeGen', 'Siq', 'SiqCache', 'SiqGen', 'SiqType', 'Snowflake', 'SnowflakeGen',
@ -51,9 +55,9 @@ __all__ = (
'WebColor', 'XYZColor', 'WebColor', 'XYZColor',
'addattr', 'additem', 'age_and_days', 'alru_cache', 'b2048decode', 'b2048encode', 'addattr', 'additem', 'age_and_days', 'alru_cache', 'b2048decode', 'b2048encode',
'b32ldecode', 'b32lencode', 'b64encode', 'b64decode', 'cb32encode', 'b32ldecode', 'b32lencode', 'b64encode', 'b64decode', 'cb32encode',
'cb32decode', 'chalk', 'count_ones', 'dei_args', 'deprecated', 'cb32decode', 'chalk', 'cooldown', 'count_ones', 'dei_args', 'deprecated', 'do_not_flood',
'future', 'ilex', 'join_bits', 'future', 'ilex', 'join_bits', 'jsonencode', 'kwargs_prefix',
'jsonencode', 'kwargs_prefix', 'lex', 'ltuple', 'makelist', 'mask_shift', 'lex', 'ltuple', 'makelist', 'mask_shift',
'matches', 'mod_ceil', 'mod_floor', 'must_be', 'none_pass', 'not_implemented', 'matches', 'mod_ceil', 'mod_floor', 'must_be', 'none_pass', 'not_implemented',
'not_less_than', 'not_greater_than', 'not_less_than', 'not_greater_than',
'redact_url_password', 'rtuple', 'split_bits', 'ssv_list', 'symbol_table', 'redact_url_password', 'rtuple', 'split_bits', 'ssv_list', 'symbol_table',

140
src/suou/argparse.py Normal file
View file

@ -0,0 +1,140 @@
"""
Utilities for parsing arguments. Based on argparse.
---
Copyright (c) 2025 Sakuragasaki46.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
See LICENSE for the specific language governing permissions and
limitations under the License.
This software is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
from __future__ import annotations
import argparse
import sys
from typing import Callable
class LetterSubparsers(object):
"""
Subparsers in pacman style, where action name can be shortened to a single letter, prefixed by a hyphen
(i.e. "-S" is expanded to "sync")
*New in 0.13.0*
"""
_parser: argparse.ArgumentParser
_letters: dict[str, str]
_subparsers: argparse._SubParsersAction[argparse.ArgumentParser]
_has_verbose: bool = False
def __init__(self, parser : argparse.ArgumentParser, *, dest: str = 'action', **kwargs):
self._parser = parser
self._letters = {}
self._subparsers = parser.add_subparsers(dest = dest, required = True, **kwargs)
def add_verbose(self, *, help: str = "show more logs (e.g. debug)"):
"""
Add a "-v" / "--verbose" argument to the main parser.
This allows the argument to be everywhere in the argv.
"""
self._parser.add_argument('-v', '--verbose', action='count', default = 0, help=help)
self._has_verbose = True
def action(self, /, letter: str, name: str | None = None, **kwargs):
"""
Decorator which adds a subparser of an argument parser's subparsers, and specifies a letter to make a shorthand in pacman style.
For example, assuming name="sync" and letter="S", if the first argument is "-S", it will be turned to "sync", .
The first argument is always the object returned by ArgumentParser.add_subparsers().
Additional kwargs are passed to the add_parser constructor.
"""
if len(letter) != 1:
raise ValueError('letter must be one character')
o_name = name
def decorator(func: Callable[argparse.ArgumentParser, ...]):
name = o_name or func.__name__
parser = self._subparsers.add_parser(name, **kwargs)
func(parser)
self._letters[letter] = name
return func
return decorator
def parse_args(self, argv: list[str] = None, system_exit: bool = True):
"""
Variation of ArgumentParser.parse_args() that takes shortcut letters into account.
Best used together with .action().
"""
if argv is None:
argv = sys.argv[1:]
else:
argv = list(argv)
opt_start = 0
# preprocess the letters
if len(argv) > 0:
first_arg = argv.pop(0)
if first_arg.startswith('-') and len(first_arg) >= 2:
for idx, letter in enumerate(first_arg):
rest = first_arg[1:idx] + first_arg[idx+1:]
if letter in self._letters:
argv.insert(0, self._letters[letter])
if rest:
argv.insert(1, "-" + rest)
opt_start = 1
break
else:
# put it back
argv.insert(0, first_arg)
else:
# put it back
argv.insert(0, first_arg)
# preprocess the verbose
if self._has_verbose and len(argv) > opt_start:
nargv = argv[:opt_start]
vc = 0
for arg in argv[opt_start:]:
if arg.startswith('-') and not arg.startswith('--'):
arg_vc = sum(1 for l in arg[1:] if l == 'v')
arg_vless = arg.replace('v', '')
if arg_vless != '-':
nargv.append(arg_vless)
vc += arg_vc
elif arg == '--verbose':
vc += 1
else:
nargv.append(arg)
argv = ["--verbose"] * vc + nargv
try:
args = self._parser.parse_args(argv)
except SystemExit:
# prevent SystemExit at parse fail
if system_exit:
raise
else:
return args
__all__ = ('LetterSubparsers',)

116
src/suou/ast.py Normal file
View file

@ -0,0 +1,116 @@
"""
Experimental AST module
---
Copyright (c) 2026 Sakuragasaki46.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
See LICENSE for the specific language governing permissions and
limitations under the License.
This software is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
from __future__ import annotations
from ast import TypeVar
from typing import Any, Collection
_T = TypeVar('_T')
class Node:
__slots__ = ('_children', )
_count: int
_children: list[Node]
_types: Collection[type]
def __repr__(self):
return f"{self.__class__.__name__}({', '.join(repr(x) for x in self._children)})"
def __init__(self, *args):
if self._count != len(args):
raise TypeError(f'{self.__class__.__name__} must be instanced with {self._count} arguments, got {len(args)}')
for i, (a, t) in enumerate(zip(args, self._types)):
if t != Any and not isinstance(a, t):
raise TypeError(f"argument {i} must be of type {t.__name__!r}, got {a.__class__.__name__!r}")
self._children = list(args)
def eval(self, ctx: dict):
raise TypeError(f'{self.__class__.__name__} cannot be eval()ed')
def __getitem__(self, key: int):
if key >= self._count:
raise IndexError('child node index out of range')
return self._children[key]
def __len__(self):
return self._count
class ZeroOp(Node):
_count = 0
_types = ()
class UnaryOp(Node):
_count = 1
_types = (Any,)
class BinaryOp(Node):
_count = 2
_types = (Any, Any)
class TernaryOp(Node):
_count = 3
_types = (Any, Any, Any)
class MultiOp(Node):
_count = 1
_types = (list,)
def __getitem__(self, key: int):
return self._children[0][key]
def __iter__(self):
return iter(self._children[0])
def __len__(self):
return len(self._children[0])
class Literal(UnaryOp):
"""A literal evals to the enclosed value."""
def eval(self, ctx: dict):
return self[0]
## SUOU provides some ready-made literals, for the sake of ease-of-use.
class IntLiteral(Literal):
_types = (int,)
class FloatLiteral(Literal):
"""
WARNING: may be subject to loss of precision.
"""
_types = (float,)
class StringLiteral(Literal):
_types = (str,)
# This module is experimental and therefore not re-exported into __init__
__all__ = (
'Node', 'ZeroOp', 'UnaryOp', 'BinaryOp', 'TernaryOp',
'MultiOp', 'Literal', 'IntLiteral', 'FloatLiteral',
'StringLiteral'
)

View file

@ -21,6 +21,7 @@ from __future__ import annotations
from collections import namedtuple from collections import namedtuple
from functools import lru_cache from functools import lru_cache
import logging
import math import math
from suou.functools import deprecated from suou.functools import deprecated
@ -331,4 +332,48 @@ class OKLCHColor(namedtuple('_OKLCHColor', 'l c h')):
return sum(abs(i - j) / k for i, j, k in zip(self, other, (1, 1, 36))) return sum(abs(i - j) / k for i, j, k in zip(self, other, (1, 1, 36)))
__all__ = ('chalk', 'WebColor', "RGBColor", 'LinearRGBColor', 'XYZColor', 'OKLabColor', 'OKLCHColor') class ColorFormatter(logging.Formatter):
"""
Colored logging formatter.
Opinionated.
Taken from https://stackoverflow.com/questions/384076/how-can-i-color-python-logging-output
"""
def _get_base_format(color):
return f"[%(asctime)s] {color('%(levelname)s')}{chalk.grey(': ')}{color('%(name)s')}{chalk.grey(': ')}%(message)s"
FORMATS = {
logging.DEBUG: _get_base_format(chalk.cyan),
logging.INFO: _get_base_format(chalk.green),
logging.WARNING: _get_base_format(chalk.yellow),
logging.ERROR: _get_base_format(chalk.red),
logging.CRITICAL: _get_base_format(chalk.bold.red)
}
del _get_base_format
def format(self, record: logging.LogRecord) -> str:
log_fmt = self.FORMATS.get(record.levelno)
formatter = logging.Formatter(log_fmt)
return formatter.format(record)
@classmethod
def apply_handler(cls, logger: logging.Logger, level = logging.INFO):
"""
Apply the colored formatter to a logger.
Use this with logging.root, instead of logging.basicConfig().
Should not be called more than once.
"""
logger.setLevel(level)
ch = logging.StreamHandler()
ch.setLevel(level)
ch.setFormatter(cls())
logger.addHandler(ch)
__all__ = ('chalk', 'WebColor', "RGBColor", 'LinearRGBColor', 'XYZColor',
'OKLabColor', 'OKLCHColor', 'ColorFormatter')

View file

@ -21,6 +21,8 @@ from __future__ import annotations
from functools import wraps from functools import wraps
from typing import Callable, Collection, TypeVar, Any from typing import Callable, Collection, TypeVar, Any
from . import deprecated
_T = TypeVar('_T') _T = TypeVar('_T')
_U = TypeVar('_U') _U = TypeVar('_U')
@ -115,6 +117,7 @@ class Pronoun(int):
@deprecated('breaks the typing system somewhat; won\'t be removed tho')
def dei_args(**renames: dict[str, str]): def dei_args(**renames: dict[str, str]):
""" """
Allow for aliases in the keyword argument names, in form alias='real_name'. Allow for aliases in the keyword argument names, in form alias='real_name'.

View file

@ -339,6 +339,75 @@ def none_pass(func: Callable[_T, _U], *args, **kwargs) -> Callable[_T, _U]:
return func(x, *args, **kwargs) return func(x, *args, **kwargs)
return wrapper return wrapper
def cooldown(unit: int, /, exception: Exception | None = None):
'''
Implement a calling cooldown for a function of procedure.
If the decorated function is called during the cooldown,
the last result is returned (or occasionally an exception).
If an exception is passed explicitly as a decorator, it is
raised upon calling during cooldown.
Otherwise, the last result is returned (or the last
exception is raised.)
*New in 0.13.0*
'''
def decorator(func: Callable[..., _U]):
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
if wrapper.timeout_until is not None and wrapper.timeout_until > now:
if exception is not None:
raise exception
elif wrapper.last_exc is not None:
raise wrapper.last_exc
else:
return wrapper.last_result
else:
wrapper.timeout_until = now + unit
try:
wrapper.last_result = func(*args, **kwargs)
except Exception as e:
wrapper.last_exc = e
raise
else:
wrapper.last_exc = None
return wrapper.last_result
wrapper.last_result: _U | None = None
wrapper.last_exc: Exception | None = None
wrapper.timeout_until: float | None = None
return wrapper
return decorator
def do_not_flood(unit = .25):
"""
Implement a calling cooldown for a function or procedure.
If the decorated function is called during the cooldown, the function
blocks before being called again.
This is blocking and uses time.sleep().
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
if wrapper.timeout_until is not None and wrapper.timeout_until > now:
wrapper.timeout_delay += unit
time.sleep(wrapper.timeout_until - now)
wrapper.timeout_until = now + wrapper.timeout_delay
else:
wrapper.timeout_delay = unit
wrapper.timeout_until = time.time() + unit
return func(*args, **kwargs)
wrapper.timeout_until = None
wrapper.timeout_delay = unit
return wrapper
return decorator
__all__ = ( __all__ = (
'deprecated', 'not_implemented', 'timed_cache', 'none_pass', 'alru_cache' 'deprecated', 'not_implemented', 'timed_cache', 'none_pass', 'alru_cache', 'cooldown', 'do_not_flood'
) )

View file

@ -22,17 +22,13 @@ from suou.classtools import MISSING
_T = TypeVar('_T') _T = TypeVar('_T')
def makelist(l: Any, wrap: bool = True) -> list | Callable[Any, list]: def _makelist_callable(l: Callable) -> Callable[..., list]:
''' @wraps(l)
Make a list out of an iterable or a single value. def wrapper(*a, **k):
return _makelist_nowrap(l(*a, **k))
return wrapper
*Changed in 0.4.0* Now supports a callable: can be used to decorate generators and turn them into lists. def _makelist_nowrap(l: Any) -> list:
Pass wrap=False to return instead the unwrapped function in a list.
*Changed in 0.11.0*: ``wrap`` argument is now no more keyword only.
'''
if callable(l) and wrap:
return wraps(l)(lambda *a, **k: makelist(l(*a, **k), wrap=False))
if isinstance(l, (str, bytes, bytearray)): if isinstance(l, (str, bytes, bytearray)):
return [l] return [l]
elif isinstance(l, Iterable): elif isinstance(l, Iterable):
@ -42,7 +38,21 @@ def makelist(l: Any, wrap: bool = True) -> list | Callable[Any, list]:
else: else:
return [l] return [l]
def ltuple(seq: Iterable[_T], size: int, /, pad = None) -> tuple: def makelist(l: Any, wrap: bool = True) -> list | Callable[..., list]:
'''
Make a list out of an iterable or a single value.
*Changed in 0.4.0* Now supports a callable: can be used to decorate generators and turn them into lists.
Pass wrap=False to return instead the unwrapped function in a list.
*Changed in 0.11.0*: ``wrap`` argument is now no more keyword only.
'''
if callable(l) and wrap:
return _makelist_callable(l)
else:
return _makelist_nowrap(l)
def ltuple(seq: Iterable[_T], size: int, /, pad = None) -> tuple[_T, ...]:
""" """
Truncate an iterable into a fixed size tuple, if necessary padding it. Truncate an iterable into a fixed size tuple, if necessary padding it.
""" """
@ -51,7 +61,7 @@ def ltuple(seq: Iterable[_T], size: int, /, pad = None) -> tuple:
seq = seq + (pad,) * (size - len(seq)) seq = seq + (pad,) * (size - len(seq))
return seq return seq
def rtuple(seq: Iterable[_T], size: int, /, pad = None) -> tuple: def rtuple(seq: Iterable[_T], size: int, /, pad = None) -> tuple[_T, ...]:
""" """
Same as rtuple() but the padding and truncation is made right to left. Same as rtuple() but the padding and truncation is made right to left.
""" """
@ -81,7 +91,7 @@ def kwargs_prefix(it: dict[str, Any], prefix: str, *, remove = True, keep_prefix
it.pop(k) it.pop(k)
return ka return ka
def additem(obj: MutableMapping, /, name: str = None): def additem(obj: MutableMapping, /, name: str | None = None):
""" """
Syntax sugar for adding a function to a mapping, immediately. Syntax sugar for adding a function to a mapping, immediately.
""" """
@ -93,7 +103,7 @@ def additem(obj: MutableMapping, /, name: str = None):
return func return func
return decorator return decorator
def addattr(obj: Any, /, name: str = None): def addattr(obj: Any, /, name: str | None = None):
""" """
Same as additem() but setting as attribute instead. Same as additem() but setting as attribute instead.
""" """

View file

@ -62,8 +62,6 @@ def symbol_table(*args: Iterable[tuple | TokenSym], whitespace: str | None = Non
yield TokenSym('[' + re.escape(whitespace) + ']+', '', discard=True) yield TokenSym('[' + re.escape(whitespace) + ']+', '', discard=True)
symbol_table: Callable[..., list]
def ilex(text: str, table: Iterable[TokenSym], *, whitespace = False): def ilex(text: str, table: Iterable[TokenSym], *, whitespace = False):
""" """
Return a text as a list of tokens, given a token table (iterable of TokenSym). Return a text as a list of tokens, given a token table (iterable of TokenSym).

View file

@ -35,7 +35,7 @@ def lucky(validators: Iterable[Callable[[_U], bool]] = ()):
*New in 0.7.0* *New in 0.7.0*
""" """
def decorator(func: Callable[_T, _U]) -> Callable[_T, _U]: def decorator(func: Callable[..., _U]) -> Callable[..., _U]:
@wraps(func) @wraps(func)
def wrapper(*args, **kwargs) -> _U: def wrapper(*args, **kwargs) -> _U:
try: try:

View file

@ -86,4 +86,5 @@ class PingExtension(markdown.extensions.Extension):
md.inlinePatterns.register(MentionPattern(re.escape(at) + r'(' + self.CHARACTERS + ')', url_prefix), 'ping_mention', 14) md.inlinePatterns.register(MentionPattern(re.escape(at) + r'(' + self.CHARACTERS + ')', url_prefix), 'ping_mention', 14)
# Optional dependency: do not import into __init__.py
__all__ = ('PingExtension', 'SpoilerExtension', 'StrikethroughExtension') __all__ = ('PingExtension', 'SpoilerExtension', 'StrikethroughExtension')

View file

@ -22,11 +22,8 @@ from itsdangerous import TimestampSigner
from itsdangerous import Signer as _Signer from itsdangerous import Signer as _Signer
from itsdangerous.encoding import int_to_bytes as _int_to_bytes from itsdangerous.encoding import int_to_bytes as _int_to_bytes
from suou.dei import dei_args from .itertools import rtuple
from suou.itertools import rtuple from .codecs import jsondecode, jsonencode, want_bytes, want_str, b64decode, b64encode
from .functools import not_implemented
from .codecs import jsondecode, jsonencode, rb64decode, want_bytes, want_str, b64decode, b64encode
from .iding import Siq from .iding import Siq
from .classtools import MISSING from .classtools import MISSING
@ -35,7 +32,6 @@ class UserSigner(TimestampSigner):
itsdangerous.TimestampSigner() instanced from a user ID, with token generation and validation capabilities. itsdangerous.TimestampSigner() instanced from a user ID, with token generation and validation capabilities.
""" """
user_id: int user_id: int
@dei_args(primary_secret='master_secret')
def __init__(self, master_secret: bytes, user_id: int, user_secret: bytes, **kwargs): def __init__(self, master_secret: bytes, user_id: int, user_secret: bytes, **kwargs):
super().__init__(master_secret + user_secret, salt=Siq(user_id).to_bytes(), **kwargs) super().__init__(master_secret + user_secret, salt=Siq(user_id).to_bytes(), **kwargs)
self.user_id = user_id self.user_id = user_id

View file

@ -154,7 +154,7 @@ def require_auth_base(cls: type[DeclarativeBase], *, src: AuthSrc, column: str |
return decorator return decorator
from .asyncio import SQLAlchemy, async_query from .asyncio import SQLAlchemy, async_query, SessionWrapper, AsyncSelectPagination
from .orm import ( from .orm import (
id_column, snowflake_column, match_column, match_constraint, bool_column, declarative_base, parent_children, id_column, snowflake_column, match_column, match_constraint, bool_column, declarative_base, parent_children,
author_pair, age_pair, bound_fk, unbound_fk, want_column, a_relationship, BitSelector, secret_column, username_column author_pair, age_pair, bound_fk, unbound_fk, want_column, a_relationship, BitSelector, secret_column, username_column
@ -168,7 +168,7 @@ except ImportError:
# Optional dependency: do not import into __init__.py # Optional dependency: do not import into __init__.py
__all__ = ( __all__ = (
'IdType', 'id_column', 'snowflake_column', 'entity_base', 'declarative_base', 'token_signer', 'IdType', 'id_column', 'snowflake_column', 'declarative_base', 'token_signer',
'match_column', 'match_constraint', 'bool_column', 'parent_children', 'match_column', 'match_constraint', 'bool_column', 'parent_children',
'author_pair', 'age_pair', 'bound_fk', 'unbound_fk', 'want_column', 'author_pair', 'age_pair', 'bound_fk', 'unbound_fk', 'want_column',
'a_relationship', 'BitSelector', 'secret_column', 'username_column', 'a_relationship', 'BitSelector', 'secret_column', 'username_column',

View file

@ -22,8 +22,8 @@ from functools import wraps
from contextvars import ContextVar, Token from contextvars import ContextVar, Token
from typing import Callable, TypeVar from typing import Callable, TypeVar
from sqlalchemy import Select, Table, func, select from sqlalchemy import Select, Table, select
from sqlalchemy.orm import DeclarativeBase, lazyload from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
try: try:
@ -31,8 +31,7 @@ try:
except ImportError: except ImportError:
AsyncSelectPagination = None AsyncSelectPagination = None
from suou.exceptions import NotFoundError from ..exceptions import NotFoundError
from suou.glue import glue
_T = TypeVar('_T') _T = TypeVar('_T')
_U = TypeVar('_U') _U = TypeVar('_U')
@ -126,9 +125,11 @@ class SQLAlchemy:
self.engine, checkfirst=checkfirst self.engine, checkfirst=checkfirst
) )
# XXX NOT public API! DO NOT USE
current_session: ContextVar[AsyncSession] = ContextVar('current_session')
current_session: ContextVar[AsyncSession] = ContextVar('current_session')
"""
XXX NOT public API! DO NOT USE
"""
def async_query(db: SQLAlchemy, multi: False): def async_query(db: SQLAlchemy, multi: False):
""" """

View file

@ -18,7 +18,6 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
from binascii import Incomplete
import os import os
import re import re
from typing import Any, Callable, TypeVar from typing import Any, Callable, TypeVar
@ -27,10 +26,8 @@ from sqlalchemy import BigInteger, Boolean, CheckConstraint, Column, Date, Forei
from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, Relationship, declarative_base as _declarative_base, relationship from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, Relationship, declarative_base as _declarative_base, relationship
from sqlalchemy.types import TypeEngine from sqlalchemy.types import TypeEngine
from sqlalchemy.ext.hybrid import Comparator from sqlalchemy.ext.hybrid import Comparator
from suou.functools import future
from suou.classtools import Wanted, Incomplete from suou.classtools import Wanted, Incomplete
from suou.codecs import StringCase from suou.codecs import StringCase
from suou.dei import dei_args
from suou.iding import Siq, SiqCache, SiqGen, SiqType from suou.iding import Siq, SiqCache, SiqGen, SiqType
from suou.itertools import kwargs_prefix from suou.itertools import kwargs_prefix
from suou.snowflake import SnowflakeGen from suou.snowflake import SnowflakeGen
@ -131,6 +128,23 @@ def username_column(
return match_column(length, regex, case=case, nullable=nullable, unique=True, *args, **kwargs) return match_column(length, regex, case=case, nullable=nullable, unique=True, *args, **kwargs)
EMAIL_RE_USERNAME = r"[a-z0-9-]+(\.[a-z0-9-]+)*"
EMAIL_RE_DOMAIN = r"([a-z0-9-]+\.)+[a-z0-9-]{2,15}"
EMAIL_RE_YESALIASES = EMAIL_RE_USERNAME + r"(\+" + EMAIL_RE_USERNAME + ")?@" + EMAIL_RE_DOMAIN
EMAIL_RE_NOALIASES = EMAIL_RE_USERNAME + r"@" + EMAIL_RE_DOMAIN
def email_column(
length: int = 256, *args, allow_aliases: bool = True, nullable: bool = False, unique: bool = True, **kwargs
):
"""
Construct a column containing a email address.
*New in 0.14.0*
"""
return match_column(length, EMAIL_RE_YESALIASES if allow_aliases else EMAIL_RE_NOALIASES, case = StringCase.FORCE_LOWER,
unique = unique, nullable = nullable, *args, **kwargs)
def bool_column(value: bool = False, nullable: bool = False, **kwargs) -> Column[bool]: def bool_column(value: bool = False, nullable: bool = False, **kwargs) -> Column[bool]:
""" """
Column for a single boolean value. Column for a single boolean value.
@ -141,11 +155,12 @@ def bool_column(value: bool = False, nullable: bool = False, **kwargs) -> Column
return Column(Boolean, server_default=def_val, nullable=nullable, **kwargs) return Column(Boolean, server_default=def_val, nullable=nullable, **kwargs)
@dei_args(primary_secret='master_secret')
def declarative_base(domain_name: str, master_secret: bytes, metadata: dict | None = None, **kwargs) -> type[DeclarativeBase]: def declarative_base(domain_name: str, master_secret: bytes, metadata: dict | None = None, **kwargs) -> type[DeclarativeBase]:
""" """
Drop-in replacement for sqlalchemy.orm.declarative_base() Drop-in replacement for sqlalchemy.orm.declarative_base()
taking in account requirements for SIQ generation (i.e. domain name). taking in account requirements for SIQ generation (i.e. domain name).
Also supports snowflake generation parameters such as epoch.
""" """
if not isinstance(metadata, dict): if not isinstance(metadata, dict):
metadata = dict() metadata = dict()
@ -160,7 +175,6 @@ def declarative_base(domain_name: str, master_secret: bytes, metadata: dict | No
) )
Base = _declarative_base(metadata=MetaData(**metadata), **kwargs) Base = _declarative_base(metadata=MetaData(**metadata), **kwargs)
return Base return Base
entity_base = warnings.deprecated('use declarative_base() instead')(declarative_base)

View file

@ -28,7 +28,7 @@ from suou.functools import future
@future() @future()
class Waiter(): class Waiter():
_cached_app: Callable | None = None _cached_app: Starlette | None = None
def __init__(self): def __init__(self):
self.routes: list[Route] = [] self.routes: list[Route] = []
@ -60,7 +60,7 @@ class Waiter():
def patch(self, endpoint: str, *a, **k): def patch(self, endpoint: str, *a, **k):
return self._route('PATCH', endpoint, *a, **k) return self._route('PATCH', endpoint, *a, **k)
def _route(self, methods: list[str], endpoint: str, **kwargs): def _route(self, methods: str | list[str], endpoint: str, **kwargs):
def decorator(func): def decorator(func):
self.routes.append(Route(endpoint, func, methods=makelist(methods, False), **kwargs)) self.routes.append(Route(endpoint, func, methods=makelist(methods, False), **kwargs))
return func return func