implement proxy inheritance mode

Author: collerek

docs/models/inheritance.md

@@ -1,9 +1,10 @@
 # Inheritance
 
-Out of various types of ORM models inheritance `ormar` currently supports two of them:
+Out of various types of ORM models inheritance `ormar` currently supports three of them:
 
 * **Mixins**
 * **Concrete table inheritance** (with parents set to `abstract=True`)
+* **Proxy models** (with children set to `proxy=True`)
 
 ## Types of inheritance
 
@@ -15,6 +16,8 @@ The short summary of different types of inheritance:
 * **Concrete table inheritance [SUPPORTED]** - means that parent is marked as abstract
   and each child has its own table with columns from a parent and own child columns, kind
   of similar to Mixins but parent also is a Model
+* **Proxy models [SUPPORTED]** - means that only parent has an actual table,
+  children just add methods, modify settings etc. and share the parent's table
 * **Single table inheritance [NOT SUPPORTED]** - means that only one table is created
   with fields that are combination/sum of the parent and all children models but child
   models use only subset of column in db (all parent and own ones, skipping the other
@@ -23,8 +26,6 @@ The short summary of different types of inheritance:
   is saved on parent model and part is saved on child model that are connected to each
   other by kind of one to one relation and under the hood you operate on two models at
   once
-* **Proxy models [NOT SUPPORTED]** - means that only parent has an actual table,
-  children just add methods, modify settings etc.
 
 ## Mixins
 
@@ -193,6 +194,68 @@ assert isinstance(
 
     `created_date: str = ormar.String(max_length=200, name="creation_date2") # exception`
 
+## Proxy models
+
+A proxy model is a child class that **shares its parent's table** and only adds
+methods, computed fields, or a custom `queryset_class`. Use it when you want
+multiple Python views over the same underlying table, for example to attach
+domain methods to a generic record type.
+
+To declare a proxy model, set `proxy=True` on the child's `OrmarConfig`:
+
+```python
+base_ormar_config = ormar.OrmarConfig(
+    database=DatabaseConnection(DATABASE_URL),
+    metadata=sqlalchemy.MetaData(),
+)
+
+
+class Human(ormar.Model):
+    ormar_config = base_ormar_config.copy(tablename="humans")
+
+    id: int = ormar.Integer(primary_key=True)
+    first_name: str = ormar.String(max_length=50)
+    last_name: str = ormar.String(max_length=50)
+
+
+class User(Human):
+    ormar_config = base_ormar_config.copy(proxy=True)
+
+    def full_name(self) -> str:
+        return f"{self.first_name} {self.last_name}"
+```
+
+Both `Human.objects.all()` and `User.objects.all()` read from the same `humans`
+table; the only difference is that `User` rows are returned as `User` instances
+and gain the `full_name()` method.
+
+### Constraints
+
+* The base class must be a non-abstract `ormar.Model`. Proxying an
+  `abstract=True` parent raises `ModelDefinitionError` — use concrete
+  inheritance for that.
+* A proxy model **cannot** declare new ormar fields. Doing so raises
+  `ModelDefinitionError` because the schema is fixed by the parent.
+* `proxy=True` and `abstract=True` are mutually exclusive on the same class.
+* Proxy chains (e.g. `Admin(User)` with `proxy=True` where `User` itself is a
+  proxy of `Human`) are allowed and resolve back to the root concrete table.
+* All concrete ormar bases of a proxy must share the same table.
+
+### Behavior notes
+
+* Each proxy class has its own `SignalEmitter`, so a `pre_save` registered on
+  `User` does not fire when a `Human` instance is saved (and vice versa).
+  Set `emit_parent_signals=True` on the proxy's `OrmarConfig` to also dispatch
+  the parent's `pre_save` / `post_save` / `pre_update` / `post_update` /
+  `pre_delete` / `post_delete` handlers (with `sender=parent_cls`) on every
+  save / update / delete via the proxy. The flag is opt-in to preserve the
+  contract of existing handlers that assume the instance is the parent type.
+* Reverse relations defined on related models still point at the original
+  parent class — proxying does not create new reverse accessors.
+* The proxy's `queryset_class` and `extra` settings can be overridden via
+  `OrmarConfig.copy(proxy=True, queryset_class=...)` independently of the
+  parent.
+
 ## Relations in inheritance
 
 You can declare relations in every step of inheritance, so both in parent and child

ormar/models/helpers/models.py

@@ -71,12 +71,20 @@ def check_required_config_parameters(new_model: type["Model"]) -> None:
     :param new_model: newly declared ormar Model
     :type new_model: Model class
     """
-    if new_model.ormar_config.database is None and not new_model.ormar_config.abstract:
+    if new_model.ormar_config.proxy and new_model.ormar_config.abstract:
+        raise ormar.ModelDefinitionError(
+            f"{new_model.__name__} cannot be both proxy and abstract."
+        )
+
+    skip_db_metadata_checks = (
+        new_model.ormar_config.abstract or new_model.ormar_config.proxy
+    )
+    if new_model.ormar_config.database is None and not skip_db_metadata_checks:
         raise ormar.ModelDefinitionError(
             f"{new_model.__name__} does not have database defined."
         )
 
-    if new_model.ormar_config.metadata is None and not new_model.ormar_config.abstract:
+    if new_model.ormar_config.metadata is None and not skip_db_metadata_checks:
         raise ormar.ModelDefinitionError(
             f"{new_model.__name__} does not have metadata defined."
         )

ormar/models/metaclass.py

@@ -353,7 +353,33 @@ def copy_data_from_parent_model(  # noqa: CCR001
     :rtype: tuple[dict, dict]
     """
     if attrs.get("ormar_config"):
-        if model_fields and not base_class.ormar_config.abstract:  # type: ignore
+        child_config = attrs["ormar_config"]
+        child_proxy = getattr(child_config, "proxy", False)
+        base_abstract = base_class.ormar_config.abstract  # type: ignore
+
+        if child_proxy:
+            if base_abstract:
+                raise ModelDefinitionError(
+                    f"Proxy model {curr_class.__name__} cannot inherit from "
+                    f"abstract class {base_class.__name__}; "
+                    f"use concrete inheritance instead."
+                )
+            new_fields = set(model_fields) - set(
+                base_class.ormar_config.model_fields  # type: ignore
+            )
+            if new_fields:
+                raise ModelDefinitionError(
+                    f"Proxy model {curr_class.__name__} cannot declare new ormar "
+                    f"fields: {sorted(new_fields)}."
+                )
+            update_attrs_from_base_config(
+                base_class=base_class,  # type: ignore
+                attrs=attrs,
+                model_fields=model_fields,
+            )
+            return attrs, dict(base_class.ormar_config.model_fields)
+
+        if model_fields and not base_abstract:
             raise ModelDefinitionError(
                 f"{curr_class.__name__} cannot inherit "
                 f"from non abstract class {base_class.__name__}"
@@ -521,6 +547,50 @@ def update_attrs_and_fields(
     return updated_model_fields
 
 
+def wire_proxy_from_parent(new_model: type["Model"]) -> None:
+    """
+    Resolve the concrete ormar parent of a proxy model and share its table,
+    columns, primary key, model_fields, metadata and database with the proxy.
+
+    Proxy models do not own a SQLAlchemy table. They reuse the parent's table
+    so that queries via the proxy class hit the same physical rows.
+
+    :raises ModelDefinitionError: if the proxy has no concrete ormar parent or
+        if it inherits from multiple concrete ormar models with different tables.
+    :param new_model: the proxy model class being constructed
+    :type new_model: type["Model"]
+    """
+    concrete_bases = [
+        base
+        for base in new_model.__mro__[1:]
+        if hasattr(base, "ormar_config")
+        and not base.ormar_config.abstract
+        and base is not new_model
+    ]
+    if not concrete_bases:
+        raise ModelDefinitionError(
+            f"Proxy model {new_model.__name__} has no concrete ormar parent."
+        )
+    primary_table = concrete_bases[0].ormar_config.table
+    for other in concrete_bases[1:]:
+        if other.ormar_config.table is not primary_table:
+            raise ModelDefinitionError(
+                f"Proxy model {new_model.__name__} cannot inherit from multiple "
+                f"concrete ormar models with different tables."
+            )
+    parent_cfg = concrete_bases[0].ormar_config
+    cfg = new_model.ormar_config
+    cfg.table = parent_cfg.table
+    cfg.tablename = parent_cfg.tablename
+    cfg.pkname = parent_cfg.pkname
+    cfg.columns = parent_cfg.columns
+    cfg.model_fields = parent_cfg.model_fields
+    cfg.metadata = parent_cfg.metadata
+    cfg.database = parent_cfg.database
+    cfg.alias_manager = parent_cfg.alias_manager
+    new_model.pk = PkDescriptor(name=parent_cfg.pkname)
+
+
 def add_field_descriptor(
     name: str, field: "BaseField", new_model: type["Model"]
 ) -> None:
@@ -653,7 +723,9 @@ def __new__(  # type: ignore # noqa: CCR001
             register_signals(new_model=new_model)
             modify_schema_example(model=new_model)
 
-            if not new_model.ormar_config.abstract:
+            if new_model.ormar_config.proxy:
+                wire_proxy_from_parent(new_model)
+            elif not new_model.ormar_config.abstract:
                 new_model = populate_config_tablename_columns_and_pk(name, new_model)
                 populate_config_sqlalchemy_table_if_required(new_model.ormar_config)
                 expand_reverse_relationships(new_model)

ormar/models/model.py

@@ -38,6 +38,32 @@ async def _execute_query(self, expr: Executable, is_select: bool = False) -> Any
             )
         return row
 
+    async def _emit_signal(self, name: str, **kwargs: Any) -> None:
+        """
+        Emit a lifecycle signal on this model's SignalEmitter.
+
+        When ``ormar_config.emit_parent_signals`` is True, the same signal is
+        also dispatched on every concrete ormar ancestor in the MRO. Each
+        emit uses ``sender=ancestor_cls`` so handlers registered with
+        ``@pre_save(Parent)`` see the parent class as the sender.
+
+        :param name: signal name on the SignalEmitter (e.g. ``"pre_save"``).
+        :type name: str
+        :param kwargs: extra payload forwarded to receivers.
+        :type kwargs: Any
+        """
+        cls = type(self)
+        await getattr(self.ormar_config.signals, name).send(sender=cls, **kwargs)
+        if not self.ormar_config.emit_parent_signals:
+            return
+        seen = {id(self.ormar_config.signals)}
+        for ancestor in cls.__mro__[1:]:
+            cfg = getattr(ancestor, "ormar_config", None)
+            if cfg is None or cfg.abstract or id(cfg.signals) in seen:
+                continue
+            seen.add(id(cfg.signals))
+            await getattr(cfg.signals, name).send(sender=ancestor, **kwargs)
+
     async def upsert(self: T, **kwargs: Any) -> T:
         """
         Performs either a save or an update depending on the presence of the pk.
@@ -85,7 +111,7 @@ async def save(self: T) -> T:
         :return: saved Model
         :rtype: Model
         """
-        await self.signals.pre_save.send(sender=self.__class__, instance=self)
+        await self._emit_signal("pre_save", instance=self)
         self_fields = self._extract_model_db_fields()
 
         if (
@@ -138,7 +164,7 @@ async def save(self: T) -> T:
             await self.load()
 
         self.__setattr_fields__.clear()
-        await self.signals.post_save.send(sender=self.__class__, instance=self)
+        await self._emit_signal("post_save", instance=self)
         return self
 
     async def save_related(  # noqa: CCR001, CFQ002
@@ -274,9 +300,7 @@ async def update(self: T, _columns: Optional[list[str]] = None, **kwargs: Any) -
                 "You cannot update not saved model! Use save or upsert method."
             )
 
-        await self.signals.pre_update.send(
-            sender=self.__class__, instance=self, passed_args=kwargs
-        )
+        await self._emit_signal("pre_update", instance=self, passed_args=kwargs)
         self_fields = self._extract_model_db_fields()
         self_fields.pop(self.get_column_name_from_alias(self.ormar_config.pkname))
         if _columns:
@@ -292,7 +316,7 @@ async def update(self: T, _columns: Optional[list[str]] = None, **kwargs: Any) -
             await self._execute_query(expr)
         self.set_save_status(True)
         self.__setattr_fields__.clear()
-        await self.signals.post_update.send(sender=self.__class__, instance=self)
+        await self._emit_signal("post_update", instance=self)
         return self
 
     async def delete(self) -> int:
@@ -310,12 +334,12 @@ async def delete(self) -> int:
         :return: number of deleted rows (for some backends)
         :rtype: int
         """
-        await self.signals.pre_delete.send(sender=self.__class__, instance=self)
+        await self._emit_signal("pre_delete", instance=self)
         expr = self.ormar_config.table.delete()
         expr = expr.where(self.pk_column == (getattr(self, self.ormar_config.pkname)))
         result = await self._execute_query(expr)
         self.set_save_status(False)
-        await self.signals.post_delete.send(sender=self.__class__, instance=self)
+        await self._emit_signal("post_delete", instance=self)
         return result
 
     async def load(self: T) -> T:

ormar/models/ormar_config.py

@@ -22,6 +22,8 @@ class OrmarConfig:
         tablename: str
         order_by: list[str]
         abstract: bool
+        proxy: bool
+        emit_parent_signals: bool
         exclude_parent_fields: list[str]
         constraints: list[ColumnCollectionConstraint]
 
@@ -33,6 +35,8 @@ def __init__(
         tablename: Optional[str] = None,
         order_by: Optional[list[str]] = None,
         abstract: bool = False,
+        proxy: bool = False,
+        emit_parent_signals: bool = False,
         queryset_class: type[QuerySet] = QuerySet,
         extra: Extra = Extra.forbid,
         constraints: Optional[list[ColumnCollectionConstraint]] = None,
@@ -52,6 +56,8 @@ def __init__(
         self.property_fields: set = set()
         self.signals: SignalEmitter = SignalEmitter()
         self.abstract = abstract
+        self.proxy = proxy
+        self.emit_parent_signals = emit_parent_signals
         self.requires_ref_update: bool = False
         self.extra = extra
         self.queryset_class = queryset_class
@@ -65,6 +71,8 @@ def copy(
         tablename: Optional[str] = None,
         order_by: Optional[list[str]] = None,
         abstract: Optional[bool] = None,
+        proxy: Optional[bool] = None,
+        emit_parent_signals: Optional[bool] = None,
         queryset_class: Optional[type[QuerySet]] = None,
         extra: Optional[Extra] = None,
         constraints: Optional[list[ColumnCollectionConstraint]] = None,
@@ -76,6 +84,12 @@ def copy(
             tablename=tablename,
             order_by=order_by,
             abstract=abstract or self.abstract,
+            proxy=proxy if proxy is not None else self.proxy,
+            emit_parent_signals=(
+                emit_parent_signals
+                if emit_parent_signals is not None
+                else self.emit_parent_signals
+            ),
             queryset_class=queryset_class or self.queryset_class,
             extra=extra or self.extra,
             constraints=constraints,