0.11.0 wrap SQLAlchemy() sessions by default, add Lawyer(), SpitText(), cb32lencode(), more Snowflake.from_*(), docstring changes

This commit is contained in:
Yusur 2025-11-28 10:21:26 +01:00
parent 855299c6d5
commit 0460062867
21 changed files with 220 additions and 91 deletions

View file

@ -1,5 +1,15 @@
# Changelog
## 0.11.0
+ **Breaking**: sessions returned by `SQLAlchemy()` are now wrapped by default. Restore original behavior by passing `wrap=False` to the constructor or to `begin()`
+ Slate unused `require_auth()` and derivatives for removal in 0.14.0
+ Add `cb32lencode()`
+ `Snowflake()`: add `.from_cb32()`, `.from_base64()`, `.from_oct()`, `.from_hex()` classmethods
+ Add `SpitText()`
+ Add `Lawyer()` with seven methods
+ Style changes to docstrings
## 0.10.2 and 0.7.11
+ fix incorrect types on `cb32decode()`

View file

@ -37,7 +37,7 @@ from .redact import redact_url_password
from .http import WantsContentType
from .color import chalk, WebColor
__version__ = "0.10.2"
__version__ = "0.11.0"
__all__ = (
'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue',

View file

@ -179,6 +179,12 @@ def cb32encode(val: bytes) -> str:
'''
return want_str(base64.b32encode(val)).translate(B32_TO_CROCKFORD)
def cb32lencode(val: bytes) -> str:
'''
Encode bytes in Crockford Base32, lowercased.
'''
return want_str(base64.b32encode(val)).translate(B32_TO_CROCKFORD).lower()
def cb32decode(val: bytes | str) -> bytes:
'''
Decode bytes from Crockford Base32.

View file

@ -28,7 +28,7 @@ class TimedDict(dict[_KT, _VT]):
"""
Dictionary where keys expire after the defined time to live, expressed in seconds.
NEW 0.5.0
*New in 0.5.0*
"""
_expires: dict[_KT, int]
_ttl: int

View file

@ -1,7 +1,7 @@
"""
Colors for coding artists
NEW 0.7.0
*New in 0.7.0*
---
@ -33,7 +33,7 @@ class Chalk:
UNTESTED
NEW 0.7.0
*New in 0.7.0*
"""
CSI = '\x1b['
RED = CSI + "31m"

View file

@ -109,9 +109,10 @@ class DictConfigSource(ConfigSource):
class ArgConfigSource(ValueSource):
"""
It assumes arguments have already been parsed
Config source that assumes arguments have already been parsed.
NEW 0.6"""
*New in 0.6.0*
"""
_ns: Namespace
def __init__(self, ns: Namespace):
super().__init__()

View file

@ -1,6 +1,8 @@
"""
Utilities for Flask-SQLAlchemy binding.
This module is deprecated and will be REMOVED in 0.14.0.
---
Copyright (c) 2025 Sakuragasaki46.
@ -50,27 +52,7 @@ class FlaskAuthSrc(AuthSrc):
@deprecated('not intuitive to use')
def require_auth(cls: type[DeclarativeBase], db: SQLAlchemy) -> Callable[Any, Callable]:
"""
Make an auth_required() decorator for Flask views.
This looks for a token in the Authorization header, validates it, loads the
appropriate object, and injects it as the user= parameter.
NOTE: the actual decorator to be used on routes is **auth_required()**,
NOT require_auth() which is the **constructor** for it.
cls is a SQLAlchemy table.
db is a flask_sqlalchemy.SQLAlchemy() binding.
Usage:
auth_required = require_auth(User, db)
@route('/admin')
@auth_required(validators=[lambda x: x.is_administrator])
def super_secret_stuff(user):
pass
NOTE: require_auth() DOES NOT work with flask_restx.
"""
def auth_required(**kwargs):
return require_auth_base(cls=cls, src=FlaskAuthSrc(db), **kwargs)

View file

@ -87,7 +87,7 @@ def future(message: str | None = None, *, version: str = None):
version= is the intended version release.
NEW 0.7.0
*New in 0.7.0*
"""
def decorator(func: Callable[_T, _U]) -> Callable[_T, _U]:
@wraps(func)
@ -135,7 +135,7 @@ def _make_alru_cache(_CacheInfo):
PSA there is no C speed up. Unlike PSL. Sorry.
NEW 0.5.0
*New in 0.5.0*
"""
# Users should only access the lru_cache through its public API:
@ -292,7 +292,7 @@ def timed_cache(ttl: int, maxsize: int = 128, typed: bool = False, *, async_: bo
Supports coroutines with async_=True.
NEW 0.5.0
*New in 0.5.0*
"""
def decorator(func: Callable[_T, _U]) -> Callable[_T, _U]:
start_time = None
@ -330,7 +330,7 @@ def none_pass(func: Callable[_T, _U], *args, **kwargs) -> Callable[_T, _U]:
Shorthand for func(x) if x is not None else None
NEW 0.5.0
*New in 0.5.0*
"""
@wraps(func)
def wrapper(x):

View file

@ -22,12 +22,14 @@ from suou.classtools import MISSING
_T = TypeVar('_T')
def makelist(l: Any, *, wrap: bool = True) -> list | Callable[Any, list]:
def makelist(l: Any, wrap: bool = True) -> list | Callable[Any, list]:
'''
Make a list out of an iterable or a single value.
NEW 0.4.0: Now supports a callable: can be used to decorate generators and turn them into lists.
*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 wraps(l)(lambda *a, **k: makelist(l(*a, **k), wrap=False))

View file

@ -18,6 +18,9 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# TODO more snippets
from .strtools import SpitText
INDEMNIFY = """
You agree to indemnify and hold harmless {0} from any and all claims, damages, liabilities, costs and expenses, including reasonable and unreasonable counsel and attorneys fees, arising out of any breach of this agreement.
"""
@ -27,7 +30,7 @@ Except as represented in this agreement, the {0} is provided “AS IS”. Other
"""
GOVERNING_LAW = """
These terms of services are governed by, and shall be interpreted in accordance with, the laws of {0}. You consent to the sole jurisdiction of {1} for all disputes between You and , and You consent to the sole application of {2} law for all such disputes.
These terms of services are governed by, and shall be interpreted in accordance with, the laws of {0}. You consent to the sole jurisdiction of {1} for all disputes between You and {2}, and You consent to the sole application of {3} law for all such disputes.
"""
ENGLISH_FIRST = """
@ -45,5 +48,51 @@ If one clause of these Terms of Service or any policy incorporated here by refer
"""
COMPLETENESS = """
These Terms, together with the other policies incorporated into them by reference, contain all the terms and conditions agreed upon by You and {{ app_name }} regarding Your use of the {{ app_name }} service. No other agreement, oral or otherwise, will be deemed to exist or to bind either of the parties to this Agreement.
"""
These Terms, together with the other policies incorporated into them by reference, contain all the terms and conditions agreed upon by You and {0} regarding Your use of the {0} service. No other agreement, oral or otherwise, will be deemed to exist or to bind either of the parties to this Agreement.
"""
class Lawyer(SpitText):
"""
A tool to ease the writing of Terms of Service for web apps.
NOT A REPLACEMENT FOR A REAL LAWYER AND NOT LEGAL ADVICE
*New in 0.11.0*
"""
def __init__(self, /,
app_name: str, domain_name: str,
company_name: str, jurisdiction: str,
country: str, country_adjective: str
):
self.app_name = app_name
self.domain_name = domain_name
self.company_name = company_name
self.jurisdiction = jurisdiction
self.country = country
self.country_adjective = country_adjective
def indemnify(self):
return self.format(INDEMNIFY, 'app_name')
def no_warranty(self):
return self.format(NO_WARRANTY, 'app_name', 'company_name')
def governing_law(self) -> str:
return self.format(GOVERNING_LAW, 'country', 'jurisdiction', 'app_name', 'country_adjective')
def english_first(self) -> str:
return ENGLISH_FIRST
def expect_updates(self) -> str:
return self.format(EXPECT_UPDATES, 'app_name')
def severability(self) -> str:
return SEVERABILITY
def completeness(self) -> str:
return self.format(COMPLETENESS, 'app_name')
# This module is experimental and therefore not re-exported into __init__
__all__ = ('Lawyer',)

View file

@ -1,7 +1,7 @@
"""
Fortune, RNG and esoterism.
NEW 0.7.0
*New in 0.7.0*
---
@ -33,7 +33,7 @@ def lucky(validators: Iterable[Callable[[_U], bool]] = ()):
UNTESTED
NEW 0.7.0
*New in 0.7.0*
"""
def decorator(func: Callable[_T, _U]) -> Callable[_T, _U]:
@wraps(func)
@ -61,7 +61,7 @@ class RngCallable(Callable, Generic[_T, _U]):
UNTESTED
NEW 0.7.0
*New in 0.7.0*
"""
def __init__(self, /, func: Callable[_T, _U] | None = None, weight: int = 1):
self._callables = []
@ -97,7 +97,7 @@ def rng_overload(prev_func: RngCallable[..., _U] | int | None, /, *, weight: int
UNTESTED
NEW 0.7.0
*New in 0.7.0*
"""
if isinstance(prev_func, int) and weight == 1:
weight, prev_func = prev_func, None

View file

@ -1,7 +1,7 @@
"""
"Security through obscurity" helpers for less sensitive logging
NEW 0.5.0
*New in 0.5.0*
---
@ -27,7 +27,7 @@ def redact_url_password(u: str) -> str:
scheme://username:password@hostname/path?query
^------^
NEW 0.5.0
*New in 0.5.0*
"""
return re.sub(r':[^@:/ ]+@', ':***@', u)

View file

@ -20,6 +20,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
from __future__ import annotations
from binascii import unhexlify
import os
from threading import Lock
import time
@ -28,7 +29,7 @@ import warnings
from .migrate import SnowflakeSiqMigrator
from .iding import SiqType
from .codecs import b32ldecode, b32lencode, b64encode, cb32encode
from .codecs import b32ldecode, b32lencode, b64encode, b64decode, cb32encode, cb32decode
from .functools import deprecated
@ -121,27 +122,46 @@ class Snowflake(int):
def to_bytes(self, length: int = 14, byteorder = "big", *, signed: bool = False) -> bytes:
return super().to_bytes(length, byteorder, signed=signed)
def to_base64(self, length: int = 9, *, strip: bool = True) -> str:
return b64encode(self.to_bytes(length), strip=strip)
def to_cb32(self)-> str:
return cb32encode(self.to_bytes(8, 'big'))
to_crockford = to_cb32
def to_hex(self) -> str:
return f'{self:x}'
def to_oct(self) -> str:
return f'{self:o}'
def to_b32l(self) -> str:
# PSA Snowflake Base32 representations are padded to 10 bytes!
if self < 0:
return '_' + Snowflake.to_b32l(-self)
return b32lencode(self.to_bytes(10, 'big')).lstrip('a')
@classmethod
def from_bytes(cls, b: bytes, byteorder = 'big', *, signed: bool = False) -> Snowflake:
if len(b) not in (8, 10):
warnings.warn('Snowflakes are exactly 8 bytes long', BytesWarning)
return super().from_bytes(b, byteorder, signed=signed)
def to_base64(self, length: int = 9, *, strip: bool = True) -> str:
return b64encode(self.to_bytes(length), strip=strip)
@classmethod
def from_base64(cls, val:str) -> Snowflake:
return Snowflake.from_bytes(b64decode(val))
def to_cb32(self)-> str:
return cb32encode(self.to_bytes(8, 'big'))
to_crockford = to_cb32
@classmethod
def from_cb32(cls, val:str) -> Snowflake:
return Snowflake.from_bytes(cb32decode(val))
def to_hex(self) -> str:
return f'{self:x}'
@classmethod
def from_hex(cls, val:str) -> Snowflake:
if val.startswith('_'):
return -cls.from_hex(val.lstrip('_'))
return Snowflake.from_bytes(unhexlify(val))
def to_oct(self) -> str:
return f'{self:o}'
@classmethod
def from_oct(cls, val:str) -> Snowflake:
if val.startswith('_'):
return -cls.from_hex(val.lstrip('_'))
return Snowflake(int(val, base=8))
def to_b32l(self) -> str:
# PSA Snowflake Base32 representations are padded to 10 bytes!
if self < 0:
return '_' + Snowflake.to_b32l(-self)
return b32lencode(self.to_bytes(10, 'big')).lstrip('a')
@classmethod
def from_b32l(cls, val: str) -> Snowflake:
if val.startswith('_'):
@ -149,6 +169,14 @@ class Snowflake(int):
return -cls.from_b32l(val.lstrip('_'))
return Snowflake.from_bytes(b32ldecode(val.rjust(16, 'a')))
def to_siq(self, domain: str, epoch: int, target_type: SiqType, **kwargs):
"""
Convenience method for conversion to SIQ.
(!) This does not check for existence! Always do the check yourself.
"""
return SnowflakeSiqMigrator(domain, epoch, **kwargs).to_siq(self, target_type)
@override
def __format__(self, opt: str, /) -> str:
try:
@ -179,15 +207,6 @@ class Snowflake(int):
def __repr__(self):
return f'{self.__class__.__name__}({super().__repr__()})'
def to_siq(self, domain: str, epoch: int, target_type: SiqType, **kwargs):
"""
Convenience method for conversion to SIQ.
(!) This does not check for existence! Always do the check yourself.
"""
return SnowflakeSiqMigrator(domain, epoch, **kwargs).to_siq(self, target_type)
__all__ = (
'Snowflake', 'SnowflakeGen'

View file

@ -85,7 +85,7 @@ def token_signer(id_attr: Column | str, secret_attr: Column | str) -> Incomplete
## (in)Utilities for use in web apps below
@deprecated('not part of the public API and not even working')
@deprecated('not part of the public API and not even working. Will be removed in 0.14.0')
class AuthSrc(metaclass=ABCMeta):
'''
AuthSrc object required for require_auth_base().
@ -113,7 +113,7 @@ class AuthSrc(metaclass=ABCMeta):
pass
@deprecated('not working and too complex to use. Will be removed in 0.9.0')
@deprecated('not working and too complex to use. Will be removed in 0.14.0')
def require_auth_base(cls: type[DeclarativeBase], *, src: AuthSrc, column: str | Column[_T] = 'id', dest: str = 'user',
required: bool = False, signed: bool = False, sig_dest: str = 'signature', validators: Callable | Iterable[Callable] | None = None):
'''

View file

@ -2,7 +2,7 @@
"""
Helpers for asynchronous use of SQLAlchemy.
NEW 0.5.0; moved to current location 0.6.0
*New in 0.5.0; moved to current location in 0.6.0*
---
@ -47,21 +47,23 @@ class SQLAlchemy:
user = (await session.execute(select(User).where(User.id == userid))).scalar()
# ...
NEW 0.5.0
*New in 0.5.0*
UPDATED 0.6.0: added wrap=True
*Changed in 0.6.0*: added wrap=True
UPDATED 0.6.1: expire_on_commit is now configurable per-SQLAlchemy();
*Changed in 0.6.1*: expire_on_commit is now configurable per-SQLAlchemy();
now sessions are stored as context variables
*Changed in 0.11.0*: sessions are now wrapped by default; turn it off by instantiating it with wrap=False
"""
base: DeclarativeBase
engine: AsyncEngine
_session_tok: list[Token[AsyncSession]]
_wrapsessions: bool
_xocommit: bool
_wrapsessions: bool | None
_xocommit: bool | None
NotFound = NotFoundError
def __init__(self, model_class: DeclarativeBase, *, expire_on_commit = False, wrap = False):
def __init__(self, model_class: DeclarativeBase, *, expire_on_commit = False, wrap = True):
self.base = model_class
self.engine = None
self._wrapsessions = wrap
@ -71,13 +73,13 @@ class SQLAlchemy:
def _ensure_engine(self):
if self.engine is None:
raise RuntimeError('database is not connected')
async def begin(self, *, expire_on_commit = None, wrap = False, **kw) -> AsyncSession:
async def begin(self, *, expire_on_commit = None, wrap = None, **kw) -> AsyncSession:
self._ensure_engine()
## XXX is it accurate?
s = AsyncSession(self.engine,
expire_on_commit=expire_on_commit if expire_on_commit is not None else self._xocommit,
**kw)
if wrap:
if (wrap if wrap is not None else self._wrapsessions):
s = SessionWrapper(s)
current_session.set(s)
return s
@ -252,5 +254,8 @@ class SessionWrapper:
"""
return getattr(self._session, key)
def __del__(self):
self._session.close()
# Optional dependency: do not import into __init__.py
__all__ = ('SQLAlchemy', 'AsyncSelectPagination', 'async_query', 'SessionWrapper')

View file

@ -1,7 +1,7 @@
"""
Utilities for SQLAlchemy; ORM
NEW 0.6.0 (moved)
*New in 0.6.0 (moved)*
---
@ -123,7 +123,7 @@ def username_column(
Username must match the given `regex` and be at most `length` characters long.
NEW 0.8.0
*New in 0.8.0*
"""
if case is StringCase.AS_IS:
warnings.warn('case sensitive usernames may lead to impersonation and unexpected behavior', UserWarning)
@ -135,7 +135,7 @@ def bool_column(value: bool = False, nullable: bool = False, **kwargs) -> Column
"""
Column for a single boolean value.
NEW in 0.4.0
*New in 0.4.0*
"""
def_val = text('true') if value else text('false')
return Column(Boolean, server_default=def_val, nullable=nullable, **kwargs)
@ -197,7 +197,7 @@ def secret_column(length: int = 64, max_length: int | None = None, gen: Callable
"""
Column filled in by default with random bits (64 by default). Useful for secrets.
NEW 0.6.0
*New in 0.6.0*
"""
max_length = max_length or length
return Column(LargeBinary(max_length), default=lambda: gen(length), nullable=nullable, **kwargs)
@ -215,7 +215,7 @@ def parent_children(keyword: str, /, *, lazy='selectin', **kwargs) -> tuple[Inco
Additional keyword arguments can be sourced with parent_ and child_ argument prefixes,
obviously.
CHANGED 0.5.0: the both relationship()s use lazy='selectin' attribute now by default.
*Changed in 0.5.0*: the both relationship()s use lazy='selectin' attribute now by default.
"""
parent_kwargs = kwargs_prefix(kwargs, 'parent_')
@ -231,7 +231,7 @@ def a_relationship(primary = None, /, j=None, *, lazy='selectin', **kwargs):
"""
Shorthand for relationship() that sets lazy='selectin' by default.
NEW 0.6.0
*New in 0.6.0*
"""
if j:
kwargs['primaryjoin'] = j
@ -246,7 +246,7 @@ def unbound_fk(target: str | Column | InstrumentedAttribute, typ: _T | None = No
If target is a string, make sure to pass the column type at typ= (default: IdType aka varbinary(16))!
NEW 0.5.0
*New in 0.5.0*
"""
if isinstance(target, (Column, InstrumentedAttribute)):
target_name = f'{target.table.name}.{target.name}'
@ -269,7 +269,7 @@ def bound_fk(target: str | Column | InstrumentedAttribute, typ: _T = None, **kwa
If target is a string, make sure to pass the column type at typ= (default: IdType aka varbinary(16))!
NEW 0.5.0
*New in 0.5.0*
"""
if isinstance(target, (Column, InstrumentedAttribute)):
target_name = f'{target.table.name}.{target.name}'
@ -288,7 +288,7 @@ class _BitComparator(Comparator):
"""
Comparator object for BitSelector()
NEW 0.6.0
*New in 0.6.0*
"""
_column: Column
_flag: int
@ -314,7 +314,7 @@ class BitSelector:
Mimicks peewee's 'BitField()' behavior, with SQLAlchemy.
NEW 0.6.0
*New in 0.6.0*
"""
_column: Column
_flag: int

View file

@ -1,7 +1,7 @@
"""
Helpers for asynchronous use of SQLAlchemy.
NEW 0.5.0; MOVED to sqlalchemy.asyncio in 0.6.0
*New in 0.5.0; moved to ``sqlalchemy.asyncio`` in 0.6.0*
---

View file

@ -46,5 +46,18 @@ class PrefixIdentifier:
def __str__(self):
return f'{self._prefix}'
class SpitText:
"""
A formatter for pre-compiled strings.
*New in 0.11.0*
"""
def format(self, templ: str, *attrs: Iterable[str]) -> str:
attrs = [getattr(self, attr, f'{{{{ {attr} }}}}') for attr in attrs]
return templ.format(*attrs).strip()
__all__ = ('PrefixIdentifier',)

View file

@ -25,7 +25,7 @@ def terminal_required(func):
"""
Requires the decorated callable to be fully connected to a terminal.
NEW 0.7.0
*New in 0.7.0*
"""
@wraps(func)
def wrapper(*a, **ka):

View file

@ -1,7 +1,7 @@
"""
Content serving API over HTTP, based on Starlette.
NEW 0.6.0
*New in 0.6.0*
---

42
tests/test_legal.py Normal file
View file

@ -0,0 +1,42 @@
import unittest
from suou.legal import Lawyer
EXPECTED_INDEMNIFY = """
You agree to indemnify and hold harmless TNT from any and all claims, damages, liabilities, costs and expenses, including reasonable and unreasonable counsel and attorneys fees, arising out of any breach of this agreement.
""".strip()
EXPECTED_GOVERNING_LAW = """
These terms of services are governed by, and shall be interpreted in accordance with, the laws of Wakanda. You consent to the sole jurisdiction of Asgard, Wakanda for all disputes between You and TNT, and You consent to the sole application of Wakandan law for all such disputes.
""".strip()
class TestLegal(unittest.TestCase):
def setUp(self) -> None:
self.lawyer = Lawyer(
app_name = "TNT",
company_name= "ACME, Ltd.",
country = "Wakanda",
domain_name= "example.com",
jurisdiction= "Asgard, Wakanda",
country_adjective= "Wakandan"
)
def tearDown(self) -> None:
...
def test_indemnify(self):
self.assertEqual(
self.lawyer.indemnify(),
EXPECTED_INDEMNIFY
)
def test_governing_law(self):
self.assertEqual(
self.lawyer.governing_law(),
EXPECTED_GOVERNING_LAW
)