add bound_fk(), unbound_fk(), TimedDict()

This commit is contained in:
Yusur 2025-07-24 09:48:01 +02:00
parent 303e9e2b2d
commit 002dbb0579
5 changed files with 148 additions and 4 deletions

View file

@ -2,7 +2,8 @@
## 0.5.0 ## 0.5.0
+ Add `timed_cache()` + `sqlalchemy`: add `unbound_fk()`, `bound_fk()`
+ Add `timed_cache()`, `TimedDict()`
+ Move obsolete stuff to `obsolete` package + Move obsolete stuff to `obsolete` package
## 0.4.0 ## 0.4.0

View file

@ -21,6 +21,7 @@ from .codecs import (StringCase, cb32encode, cb32decode, b32lencode, b32ldecode,
jsonencode, want_bytes, want_str, ssv_list, want_urlsafe) jsonencode, want_bytes, want_str, ssv_list, want_urlsafe)
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 .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 .functools import deprecated, not_implemented, timed_cache from .functools import deprecated, not_implemented, timed_cache
from .classtools import Wanted, Incomplete from .classtools import Wanted, Incomplete
from .itertools import makelist, kwargs_prefix, ltuple, rtuple, additem from .itertools import makelist, kwargs_prefix, ltuple, rtuple, additem
@ -35,7 +36,7 @@ __all__ = (
'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue', 'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue',
'DictConfigSource', 'EnvConfigSource', 'I18n', 'Incomplete', 'JsonI18n', 'DictConfigSource', 'EnvConfigSource', 'I18n', 'Incomplete', 'JsonI18n',
'MissingConfigError', 'MissingConfigWarning', 'PrefixIdentifier', 'Siq', 'SiqCache', 'SiqGen', 'MissingConfigError', 'MissingConfigWarning', 'PrefixIdentifier', 'Siq', 'SiqCache', 'SiqGen',
'SiqType', 'Snowflake', 'SnowflakeGen', 'StringCase', 'TomlI18n', 'Wanted', 'SiqType', 'Snowflake', 'SnowflakeGen', 'StringCase', 'TimedDict', 'TomlI18n', 'Wanted',
'additem', 'b2048decode', 'b2048encode', 'b32ldecode', 'b32lencode', 'additem', 'b2048decode', 'b2048encode', 'b32ldecode', 'b32lencode',
'b64encode', 'b64decode', 'cb32encode', 'cb32decode', 'count_ones', 'b64encode', 'b64decode', 'cb32encode', 'cb32decode', 'count_ones',
'deprecated', 'ilex', 'join_bits', 'jsonencode', 'kwargs_prefix', 'lex', 'deprecated', 'ilex', 'join_bits', 'jsonencode', 'kwargs_prefix', 'lex',

71
src/suou/collections.py Normal file
View file

@ -0,0 +1,71 @@
"""
Miscellaneous iterables
---
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 time
from typing import TypeVar
_KT = TypeVar('_KT')
class TimedDict(dict):
_expires: dict[_KT, int]
_ttl: int
def __init__(self, ttl: int, /, *args, **kwargs):
super().__init__(*args, **kwargs)
self._ttl = ttl
self._expires = dict()
def check_ex(self, key):
if super().__contains__(key):
ex = self._expires[key]
now = int(time.time())
if ex < now:
del self._expires[key]
super().__delitem__(key)
elif key in self._expires:
del self._expires[key]
def __getitem__(self, key: _KT, /):
self.check_ex(key)
return super().__getitem__(key)
def get(self, key, default=None, /):
self.check_ex(key)
return super().get(key)
def __setitem__(self, key: _KT, value, /) -> None:
self._expires = int(time.time() + self._ttl)
super().__setitem__(key, value)
def setdefault(self, key, default, /):
self.check_ex(key)
self._expires = int(time.time() + self._ttl)
return super().setdefault(key, default)
def __delitem__(self, key, /):
del self._expires[key]
super().__delitem__(key)
def __iter__(self):
for k in super():
self.check_ex(k)
return super().__iter__()
__all__ = ('TimedDict',)

31
src/suou/legal.py Normal file
View file

@ -0,0 +1,31 @@
"""
TOS / policy building blocks for the lazy.
XXX DANGER! This is not replacement for legal advice. Contact your lawyer.
---
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.
"""
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.
"""
NO_WARRANTY = """
Except as represented in this agreement, the {0} is provided AS IS. Other than as provided in this agreement, {1} makes no other warranties, express or implied, and hereby disclaims all implied warranties, including any warranty of merchantability and warranty of fitness for a particular purpose.
"""
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.
"""

View file

@ -21,7 +21,8 @@ from functools import wraps
from typing import Callable, Iterable, Never, TypeVar from typing import Callable, Iterable, Never, TypeVar
import warnings import warnings
from sqlalchemy import BigInteger, Boolean, CheckConstraint, Date, Dialect, ForeignKey, LargeBinary, Column, MetaData, SmallInteger, String, create_engine, select, text from sqlalchemy import BigInteger, Boolean, CheckConstraint, Date, Dialect, ForeignKey, LargeBinary, Column, MetaData, SmallInteger, String, create_engine, select, text
from sqlalchemy.orm import DeclarativeBase, Session, declarative_base as _declarative_base, relationship from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, Session, declarative_base as _declarative_base, relationship
from sqlalchemy.types import TypeEngine
from .snowflake import SnowflakeGen from .snowflake import SnowflakeGen
from .itertools import kwargs_prefix, makelist from .itertools import kwargs_prefix, makelist
@ -223,6 +224,44 @@ def parent_children(keyword: str, /, **kwargs):
return parent, child return parent, child
def unbound_fk(target: str | Column | InstrumentedAttribute, typ: TypeEngine | None = None, **kwargs):
"""
Shorthand for creating a "unbound" foreign key column from a column name, the referenced column.
"Unbound" foreign keys are nullable and set to null when referenced object is deleted.
If target is a string, make sure to pass the column type at typ= (default: IdType aka varbinary(16))!
"""
if isinstance(target, (Column, InstrumentedAttribute)):
target_name = f'{target.table.name}.{target.name}'
typ = target.type
elif isinstance(target, str):
target_name = target
if typ is None:
typ = IdType
return Column(typ, ForeignKey(target_name, ondelete='SET NULL'), nullable=True, **kwargs)
def bound_fk(target: str | Column | InstrumentedAttribute, typ: TypeEngine | None = None, **kwargs):
"""
Shorthand for creating a "bound" foreign key column from a column name, the referenced column.
"Bound" foreign keys are not nullable and cascade when referenced object is deleted. It means,
parent deleted -> all children deleted.
If target is a string, make sure to pass the column type at typ= (default: IdType aka varbinary(16))!
"""
if isinstance(target, (Column, InstrumentedAttribute)):
target_name = f'{target.table.name}.{target.name}'
typ = target.type
elif isinstance(target, str):
target_name = target
if typ is None:
typ = IdType
return Column(typ, ForeignKey(target_name, ondelete='CASCADE'), nullable=False, **kwargs)
def want_column(cls: type[DeclarativeBase], col: Column[_T] | str) -> Column[_T]: def want_column(cls: type[DeclarativeBase], col: Column[_T] | str) -> Column[_T]:
""" """
Return a table's column given its name. Return a table's column given its name.
@ -238,6 +277,7 @@ def want_column(cls: type[DeclarativeBase], col: Column[_T] | str) -> Column[_T]
else: else:
raise TypeError raise TypeError
## Utilities for use in web apps below
class AuthSrc(metaclass=ABCMeta): class AuthSrc(metaclass=ABCMeta):
''' '''
@ -308,5 +348,5 @@ def require_auth_base(cls: type[DeclarativeBase], *, src: AuthSrc, column: str |
# Optional dependency: do not import into __init__.py # Optional dependency: do not import into __init__.py
__all__ = ( __all__ = (
'IdType', 'id_column', 'entity_base', 'declarative_base', 'token_signer', 'match_column', 'match_constraint', 'IdType', 'id_column', 'entity_base', 'declarative_base', 'token_signer', 'match_column', 'match_constraint',
'author_pair', 'age_pair', 'require_auth_base', 'want_column' 'author_pair', 'age_pair', 'require_auth_base', 'bound_fk', 'unbound_fk', 'want_column'
) )