Compare commits

..

No commits in common. "master" and "v0.12.6" have entirely different histories.

20 changed files with 58 additions and 478 deletions

1
.gitignore vendored
View file

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

View file

@ -1,26 +1,5 @@
# 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
+ Added unittests to `dei_args()`

View file

@ -1,6 +1,6 @@
# SIS Unified Object Underarmor
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.
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.
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))
@ -15,13 +15,13 @@ It provides utilities such as:
**Python 3.10**+ with Pip is required.
```bash
$ pip install suou
$ pip install sakuragasaki46-suou
```
To install optional dependencies (i.e. `sqlalchemy`) for development use:
```bash
$ pip install suou[sqlalchemy]
$ pip install sakuragasaki46-suou[sqlalchemy]
```
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
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.
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.
As a consequence, 'add this add that' stuff is best-effort.
@ -42,6 +42,8 @@ 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.
**DO NOT ASK TO MAKE SUOU SAFE FOR CHILDREN**. Enjoy having your fingers cut.
### "LTS"
The following versions are supported: the latest, the second-to-latest, 0.12.x and 0.7.x.
@ -52,7 +54,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.
I (yusurko) may NOT be held accountable for Your use of my code.
I (sakuragasaki46) 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.

View file

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

View file

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

View file

@ -1,140 +0,0 @@
"""
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',)

View file

@ -1,116 +0,0 @@
"""
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,7 +21,6 @@ from __future__ import annotations
from collections import namedtuple
from functools import lru_cache
import logging
import math
from suou.functools import deprecated
@ -332,48 +331,4 @@ class OKLCHColor(namedtuple('_OKLCHColor', 'l c h')):
return sum(abs(i - j) / k for i, j, k in zip(self, other, (1, 1, 36)))
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')
__all__ = ('chalk', 'WebColor', "RGBColor", 'LinearRGBColor', 'XYZColor', 'OKLabColor', 'OKLCHColor')

View file

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

View file

@ -339,75 +339,6 @@ def none_pass(func: Callable[_T, _U], *args, **kwargs) -> Callable[_T, _U]:
return func(x, *args, **kwargs)
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__ = (
'deprecated', 'not_implemented', 'timed_cache', 'none_pass', 'alru_cache', 'cooldown', 'do_not_flood'
'deprecated', 'not_implemented', 'timed_cache', 'none_pass', 'alru_cache'
)

View file

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

View file

@ -62,6 +62,8 @@ def symbol_table(*args: Iterable[tuple | TokenSym], whitespace: str | None = Non
yield TokenSym('[' + re.escape(whitespace) + ']+', '', discard=True)
symbol_table: Callable[..., list]
def ilex(text: str, table: Iterable[TokenSym], *, whitespace = False):
"""
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*
"""
def decorator(func: Callable[..., _U]) -> Callable[..., _U]:
def decorator(func: Callable[_T, _U]) -> Callable[_T, _U]:
@wraps(func)
def wrapper(*args, **kwargs) -> _U:
try:

View file

@ -86,5 +86,4 @@ class PingExtension(markdown.extensions.Extension):
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')

View file

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

View file

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

View file

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

View file

@ -18,6 +18,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
from binascii import Incomplete
import os
import re
from typing import Any, Callable, TypeVar
@ -26,8 +27,10 @@ from sqlalchemy import BigInteger, Boolean, CheckConstraint, Column, Date, Forei
from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, Relationship, declarative_base as _declarative_base, relationship
from sqlalchemy.types import TypeEngine
from sqlalchemy.ext.hybrid import Comparator
from suou.functools import future
from suou.classtools import Wanted, Incomplete
from suou.codecs import StringCase
from suou.dei import dei_args
from suou.iding import Siq, SiqCache, SiqGen, SiqType
from suou.itertools import kwargs_prefix
from suou.snowflake import SnowflakeGen
@ -128,23 +131,6 @@ def username_column(
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]:
"""
Column for a single boolean value.
@ -155,12 +141,11 @@ def bool_column(value: bool = False, nullable: bool = False, **kwargs) -> Column
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]:
"""
Drop-in replacement for sqlalchemy.orm.declarative_base()
taking in account requirements for SIQ generation (i.e. domain name).
Also supports snowflake generation parameters such as epoch.
"""
if not isinstance(metadata, dict):
metadata = dict()
@ -175,6 +160,7 @@ def declarative_base(domain_name: str, master_secret: bytes, metadata: dict | No
)
Base = _declarative_base(metadata=MetaData(**metadata), **kwargs)
return Base
entity_base = warnings.deprecated('use declarative_base() instead')(declarative_base)

View file

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