attempt fixing types
This commit is contained in:
parent
a127c88159
commit
c860d9ffe1
3 changed files with 24 additions and 15 deletions
|
|
@ -34,7 +34,7 @@ from .validators import matches
|
||||||
from .redact import redact_url_password
|
from .redact import redact_url_password
|
||||||
from .http import WantsContentType
|
from .http import WantsContentType
|
||||||
|
|
||||||
__version__ = "0.5.2"
|
__version__ = "0.5.3-dev34"
|
||||||
|
|
||||||
__all__ = (
|
__all__ = (
|
||||||
'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue',
|
'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue',
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from abc import abstractmethod
|
from abc import abstractmethod
|
||||||
|
from types import EllipsisType
|
||||||
from typing import Any, Callable, Generic, Iterable, Mapping, TypeVar
|
from typing import Any, Callable, Generic, Iterable, Mapping, TypeVar
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
|
@ -24,7 +25,10 @@ _T = TypeVar('_T')
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
MISSING = object()
|
class MissingType(object):
|
||||||
|
__slots__ = ()
|
||||||
|
|
||||||
|
MISSING = MissingType()
|
||||||
|
|
||||||
def _not_missing(v) -> bool:
|
def _not_missing(v) -> bool:
|
||||||
return v and v is not MISSING
|
return v and v is not MISSING
|
||||||
|
|
@ -43,10 +47,10 @@ class Wanted(Generic[_T]):
|
||||||
Owner class will call .__set_name__() on the parent Incomplete instance;
|
Owner class will call .__set_name__() on the parent Incomplete instance;
|
||||||
the __set_name__ parameters (owner class and name) will be passed down here.
|
the __set_name__ parameters (owner class and name) will be passed down here.
|
||||||
"""
|
"""
|
||||||
_target: Callable | str | None | Ellipsis
|
_target: Callable | str | None | EllipsisType
|
||||||
def __init__(self, getter: Callable | str | None | Ellipsis):
|
def __init__(self, getter: Callable | str | None | EllipsisType):
|
||||||
self._target = getter
|
self._target = getter
|
||||||
def __call__(self, owner: type, name: str | None = None) -> _T:
|
def __call__(self, owner: type, name: str | None = None) -> _T | str | None:
|
||||||
if self._target is None or self._target is Ellipsis:
|
if self._target is None or self._target is Ellipsis:
|
||||||
return name
|
return name
|
||||||
elif isinstance(self._target, str):
|
elif isinstance(self._target, str):
|
||||||
|
|
@ -67,10 +71,10 @@ class Incomplete(Generic[_T]):
|
||||||
Missing arguments must be passed in the appropriate positions
|
Missing arguments must be passed in the appropriate positions
|
||||||
(positional or keyword) as a Wanted() object.
|
(positional or keyword) as a Wanted() object.
|
||||||
"""
|
"""
|
||||||
_obj = Callable[Any, _T]
|
_obj: Callable[..., _T]
|
||||||
_args: Iterable
|
_args: Iterable
|
||||||
_kwargs: dict
|
_kwargs: dict
|
||||||
def __init__(self, obj: Callable[Any, _T] | Wanted, *args, **kwargs):
|
def __init__(self, obj: Callable[..., _T] | Wanted, *args, **kwargs):
|
||||||
if isinstance(obj, Wanted):
|
if isinstance(obj, Wanted):
|
||||||
self._obj = lambda x: x
|
self._obj = lambda x: x
|
||||||
self._args = (obj, )
|
self._args = (obj, )
|
||||||
|
|
@ -120,7 +124,7 @@ class ValueSource(Mapping):
|
||||||
class ValueProperty(Generic[_T]):
|
class ValueProperty(Generic[_T]):
|
||||||
_name: str | None
|
_name: str | None
|
||||||
_srcs: dict[str, str]
|
_srcs: dict[str, str]
|
||||||
_val: Any | MISSING
|
_val: Any | MissingType
|
||||||
_default: Any | None
|
_default: Any | None
|
||||||
_cast: Callable | None
|
_cast: Callable | None
|
||||||
_required: bool
|
_required: bool
|
||||||
|
|
|
||||||
|
|
@ -18,10 +18,11 @@ from __future__ import annotations
|
||||||
|
|
||||||
from abc import ABCMeta, abstractmethod
|
from abc import ABCMeta, abstractmethod
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
from typing import Callable, Iterable, Never, TypeVar
|
from typing import Any, 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, InstrumentedAttribute, Relationship, Session, declarative_base as _declarative_base, relationship
|
from sqlalchemy.orm import DeclarativeBase, InstrumentedAttribute, Relationship, 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
|
||||||
|
|
@ -35,7 +36,7 @@ _T = TypeVar('_T')
|
||||||
|
|
||||||
# SIQs are 14 bytes long. Storage is padded for alignment
|
# SIQs are 14 bytes long. Storage is padded for alignment
|
||||||
# Not to be confused with SiqType.
|
# Not to be confused with SiqType.
|
||||||
IdType: type[LargeBinary] = LargeBinary(16)
|
IdType: TypeEngine = LargeBinary(16)
|
||||||
|
|
||||||
@not_implemented
|
@not_implemented
|
||||||
def sql_escape(s: str, /, dialect: Dialect) -> str:
|
def sql_escape(s: str, /, dialect: Dialect) -> str:
|
||||||
|
|
@ -158,6 +159,7 @@ def token_signer(id_attr: Column | str, secret_attr: Column | str) -> Incomplete
|
||||||
Requires a master secret (taken from Base.metadata), a user id (visible in the token)
|
Requires a master secret (taken from Base.metadata), a user id (visible in the token)
|
||||||
and a user secret.
|
and a user secret.
|
||||||
"""
|
"""
|
||||||
|
id_val: Column | Wanted[Column]
|
||||||
if isinstance(id_attr, Column):
|
if isinstance(id_attr, Column):
|
||||||
id_val = id_attr
|
id_val = id_attr
|
||||||
elif isinstance(id_attr, str):
|
elif isinstance(id_attr, str):
|
||||||
|
|
@ -168,13 +170,16 @@ def token_signer(id_attr: Column | str, secret_attr: Column | str) -> Incomplete
|
||||||
secret_val = Wanted(secret_attr)
|
secret_val = Wanted(secret_attr)
|
||||||
def token_signer_factory(owner: DeclarativeBase, name: str):
|
def token_signer_factory(owner: DeclarativeBase, name: str):
|
||||||
def my_signer(self):
|
def my_signer(self):
|
||||||
return UserSigner(owner.metadata.info['secret_key'], id_val.__get__(self, owner), secret_val.__get__(self, owner))
|
return UserSigner(
|
||||||
|
owner.metadata.info['secret_key'],
|
||||||
|
id_val.__get__(self, owner), secret_val.__get__(self, owner) # pyright: ignore[reportAttributeAccessIssue]
|
||||||
|
)
|
||||||
my_signer.__name__ = name
|
my_signer.__name__ = name
|
||||||
return my_signer
|
return my_signer
|
||||||
return Incomplete(Wanted(token_signer_factory))
|
return Incomplete(Wanted(token_signer_factory))
|
||||||
|
|
||||||
|
|
||||||
def author_pair(fk_name: str, *, id_type: type = IdType, sig_type: type | None = None, nullable: bool = False, sig_length: int | None = 2048, **ka) -> tuple[Column, Column]:
|
def author_pair(fk_name: str, *, id_type: type | TypeEngine = IdType, sig_type: type | None = None, nullable: bool = False, sig_length: int | None = 2048, **ka) -> tuple[Column, Column]:
|
||||||
"""
|
"""
|
||||||
Return an owner ID/signature column pair, for authenticated values.
|
Return an owner ID/signature column pair, for authenticated values.
|
||||||
"""
|
"""
|
||||||
|
|
@ -203,7 +208,7 @@ def age_pair(*, nullable: bool = False, **ka) -> tuple[Column, Column]:
|
||||||
return (date_col, acc_col)
|
return (date_col, acc_col)
|
||||||
|
|
||||||
|
|
||||||
def parent_children(keyword: str, /, *, lazy='selectin', **kwargs) -> tuple[Incomplete[Relationship], Incomplete[Relationship]]:
|
def parent_children(keyword: str, /, *, lazy='selectin', **kwargs) -> tuple[Incomplete[Relationship[Any]], Incomplete[Relationship[Any]]]:
|
||||||
"""
|
"""
|
||||||
Self-referential one-to-many relationship pair.
|
Self-referential one-to-many relationship pair.
|
||||||
Parent comes first, children come later.
|
Parent comes first, children come later.
|
||||||
|
|
@ -220,8 +225,8 @@ def parent_children(keyword: str, /, *, lazy='selectin', **kwargs) -> tuple[Inco
|
||||||
parent_kwargs = kwargs_prefix(kwargs, 'parent_')
|
parent_kwargs = kwargs_prefix(kwargs, 'parent_')
|
||||||
child_kwargs = kwargs_prefix(kwargs, 'child_')
|
child_kwargs = kwargs_prefix(kwargs, 'child_')
|
||||||
|
|
||||||
parent = Incomplete(relationship, Wanted(lambda o, n: o.__name__), back_populates=f'child_{keyword}s', lazy=lazy, **parent_kwargs)
|
parent: Incomplete[Relationship[Any]] = Incomplete(relationship, Wanted(lambda o, n: o.__name__), back_populates=f'child_{keyword}s', lazy=lazy, **parent_kwargs)
|
||||||
child = Incomplete(relationship, Wanted(lambda o, n: o.__name__), back_populates=f'parent_{keyword}', lazy=lazy, **child_kwargs)
|
child: Incomplete[Relationship[Any]] = Incomplete(relationship, Wanted(lambda o, n: o.__name__), back_populates=f'parent_{keyword}', lazy=lazy, **child_kwargs)
|
||||||
|
|
||||||
return parent, child
|
return parent, child
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue