Verified Commit 1d577417 authored by Manuel Vazquez Acosta's avatar Manuel Vazquez Acosta
Browse files

Implement and document the Registry and registry factories.

parent e0af9f44
Pipeline #31627 failed with stages
in 41 seconds
.. glossary::
Model
This term is usually employed only to state the persistent-related
aspects of your broader semantical model. The persistent aspect of
an entity is usually called a model. Many times there's a one-to-one
relation between your models and the tables in a DB.
Gravity doesn't enforce any type of persistency, but she does include a
bunch of integrations for common persistency implementations.
A `model` can be completely defined by different source-model classes.
This allows you split your model into parts that are more loosely
related but that you might still need to persist in a single unit.
Model Definition Class
Each of the source-level classes that make up a complete `model`:term:.
A `model`:term: can be defined by several classes that concern
themselves with an aspect of the final model.
Example::
>>> class Person(Model, Addressable, name='person'): # doctest: +SKIP
... pass
>>> class PersonWithFriends(Model, name='person'): # doctest: +SKIP
... ...
If we ever `register
<xotl.gravity.core.registry.NonStrictRegistryFactory>`:class: both
classes we'll get a single `final model class`:term: indexed by the
name 'person'. That final class will have both model definition
classes in its `MRO`:term:.
......@@ -23,11 +23,24 @@ python_requires = >=3.8
install_requires =
xotl.tools>=2.2.1,<2.3
decorator>=5.1.0,<6
immutables~=0.16
[flake8]
exclude = .tox,.git,*/migrations/*,node_modules/*,tests/*,docs/*,build/*
select = E,W,B950
ignore = E402,E501,E731,E741,W504,W503,B011,E203
ignore =
E402
E501
E731
E741
W504
W503
B011
E203
# If I have multiple sentences in a single line, that's because I
# opted-out black formatting there.
E701
max-line-length = 80
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# Copyright (c) Manuel Vázquez Acosta
# All rights reserved.
#
# This is free software; you can do what the LICENSE file allows you to.
#
from xotl.gravity.contrib.models import Model, AbstractModel
# fmt: off
class Addressable(AbstractModel): pass
class Person(Model, Addressable, name="person"): pass
class PersonWithFriends(Model, name="person"): pass
# fmt: on
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# Copyright (c) Manuel Vázquez Acosta
# All rights reserved.
#
# This is free software; you can do what the LICENSE file allows you to.
#
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# Copyright (c) Manuel Vázquez Acosta
# All rights reserved.
#
# This is free software; you can do what the LICENSE file allows you to.
#
import typing as t
from dataclasses import dataclasses
@dataclass
class DefinitionMeta:
name: str
def resolve_class_definition_name(cls, *, default):
"""Resolve the shared name of a definition class.
.. rubric:: Model Definition Classes
"""
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# Copyright (c) Manuel Vázquez Acosta
# All rights reserved.
#
# This is free software; you can do what the LICENSE file allows you to.
#
"""Basic DB model definitions.
"""
import typing as t
from dataclasses import dataclasses
from .base import DefinitionMeta, resolve_class_definition_name
@dataclass
class ModelMeta(DefinitionMeta):
abstract: bool = False
table: str = None
tablespace: str = None
class BaseModel:
_meta: t.ClassVar[ModelMeta]
def __init_subclass__(
cls,
name: str,
abstract: bool,
table: str = None,
tablespace: str = None,
**kwargs,
):
cls._meta = ModelMeta(
name,
abstract,
table,
tablespace,
)
return super().__init_subclass__(**kwargs)
class Model(BaseModel):
"""Base class for `Model Definition Classes`:term:."""
def __init_subclass__(
cls,
name: str = None,
table: str = None,
tablespace: str = None,
**kwargs,
):
if name is None:
name = (resolve_class_definition_name(cls, default=cls.__name__),)
return super().__init_subclass__(
name=name,
abstract=False,
table=table,
tablespace=tablespace,
**kwargs,
)
class AbstractModel(Model):
def __init_subclass__(
cls,
name: str = None,
table: str = None,
tablespace: str = None,
**kwargs,
):
if name is None:
name = (resolve_class_definition_name(cls, default=cls.__name__),)
return super().__init_subclass__(
name=name,
abstract=True,
table=table,
tablespace=tablespace,
**kwargs,
)
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# Copyright (c) Manuel Vázquez Acosta
# All rights reserved.
#
# This is free software; you can do what the LICENSE file allows you to.
#
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# Copyright (c) Manuel Vázquez Acosta
# All rights reserved.
#
# This is free software; you can do what the LICENSE file allows you to.
#
from __future__ import annotations
import typing as t
from dataclasses import dataclass, field
import immutables
Base = t.TypeVar("Base", bound=t.Type)
@dataclass(init=False)
class Registry(t.Generic[Base]):
"""A registry for classes of an opaque Base."""
_classes: t.Mapping[str, Base]
instance_init_args: t.Optional[t.Sequence[t.Any]] = None
instance_init_kwargs: t.Optional[t.Mapping[str, t.Any]] = None
def __init__(
self,
classes: t.Union[
t.Mapping[str, Base],
t.Sequence[t.Tuple[str, Base]],
],
instance_init_args: t.Sequence[t.Any],
instance_init_kwargs: t.Mapping[str, t.Any],
) -> None:
self._classes = immutables.Map(classes)
self.instance_init_args = instance_init_args
self.instance_init_kwargs = instance_init_kwargs
@property
def classes(self) -> t.Mapping[str, Base]:
pass
RF = t.TypeVar("RF", bound="_RegistryFactory")
@dataclass
class _RegistryFactory(t.Generic[Base]):
classes: dict[str, Base]
class_init_kwargs: t.Mapping[str, t.Any]
instance_init_args: t.Sequence[t.Any]
instance_init_kwargs: t.Mapping[str, t.Any]
def __init__(
self,
class_init_kwargs: t.Mapping[str, t.Any] = None,
instance_init_args: t.Sequence[t.Any] = None,
instance_init_kwargs: t.Mapping[str, t.Any] = None,
) -> None:
self.classes = {}
self.class_init_kwargs = dict(class_init_kwargs or {})
self.instance_init_args = tuple(instance_init_args or ())
self.instance_init_kwargs = dict(instance_init_kwargs or {})
self._registry: t.Optional[Registry[Base]] = None
def registry(self) -> Registry:
if self._registry is None:
raise AttributeError(
"Non-finalized registry factories have no 'registry'."
)
return self._registry
def __enter__(self: RF) -> RF:
if self._registry is not None:
raise ClosedRegistryFactoryError(
"Registry Factory was closed already"
)
return self
def __exit__(self, *args):
self.close()
def close(self) -> None:
self._registry = self._finalize_registry()
def _finalize_registry(self: _RegistryFactory[Base]) -> Registry[Base]:
raise NotImplementedError
@dataclass
class StrictRegistryFactory(_RegistryFactory[Base]):
"""A `Registry` factory to interatively build subclasses of classes `Base`.
This factory is strict because it requires that every *final class* has a
single leaf `Model Definition Class`:term: before producing the registry.
Trying to produce `registry <Registry>`:class: before all names are fully
finalized raises a TypeError (specifically `NonFinalClassTypeError`:class:).
If your system cannot impose this restriction, then you can use the weaker
`NonStrictRegistryFactory`:class:.
"""
RegistryFactory = StrictRegistryFactory # alias
@dataclass
class NonStrictRegistryFactory(_RegistryFactory[Base]):
pass
class RegistryTypeError(TypeError):
pass
class ClosedRegistryFactoryError(RegistryTypeError):
pass
class NonFinalClassError(RegistryTypeError):
pass
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment