Feature: SQL view class

[merlinz01]

Is your feature request related to a problem? Please describe.

I want to create a SQL view but there's no builtin way to do that with Tortoise.

Describe the solution you'd like

from tortoise import fields
from tortoise.models import Model
from tortoise.views import View


class Customer(Model):
    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=255)
    active = fields.BooleanField()


class Order(Model):
    id = fields.IntField(primary_key=True)
    customer_id: fields.IntField
    customer: fields.ForeignKeyRelation[Customer] = fields.ForeignKeyField(
        "models.Customer",
        related_name="orders",
        on_delete="RESTRICT",
    )


class CustomerOrder(View):
    # Using the same field functions as for models
    order_id = fields.IntField(source_field="id")
    # Use the double underscore separator to do joins to other tables
    customer_id = fields.IntField(source_field="customer__id")
    customer_name = fields.CharField(source_field="customer__name")

    class Meta:
        table = "customerorder"
        # Specify the source table + filter for the data
        source = Order.filter(customer__active=True)

async def get_order_info(customer_id: int) -> list[CustomerOrder]:
    return await CustomerOrder.filter(customer_id=customer_id)

which generates SQL like this:

CREATE VIEW "customerorder" AS
SELECT
    "order"."id" AS "order_id",
    "customer"."id" AS "customer_id",
    "customer"."name" AS "customer_name"
FROM "order"
JOIN "customer"
    ON "order"."customer_id" = "customer"."id"
WHERE "customer"."active" = TRUE;

and

SELECT
    "order_id",
    "customer_id",
    "customer_name"
FROM "customerorder"
WHERE "customer_id" = $1;

Describe alternatives you've considered

Writing SQL manually, but that's what an ORM is for.

Additional context

I understand the basic view SQL syntax is consistent across database implementations.

[waketzheng]

Something like this:

import sys
from typing import Any

if sys.version_info >= (3, 11):
    from typing import Self
else:
    from typing_extensions import Self

from pydantic import BaseModel

from tortoise import Model


class View(BaseModel):
    class Meta:
        model: type[Model]

    @classmethod
    async def filter(cls, **kw: Any) -> list[Self]:
        keys: dict[str, str] = {
            k: v.metadata and v.metadata[0] or k for k, v in cls.model_fields.items()
        }
        query: dict[str, Any] = {keys.get(k, k): v for k, v in kw.items()}
        qs = cls.Meta.model.filter(**query)
        field_parts: dict[str, list[str]] = {f: f.split("__") for f in keys.values()}
        if related := [ps[0] for ps in field_parts.values() if len(ps) > 1]:
            qs = qs.prefetch_related(*related)
        results: list[Self] = []
        async for obj in qs:
            d: dict[str, Any] = dict(keys)
            for k, v in keys.items():
                parts: list[str] = field_parts[v]
                value = getattr(obj, parts[0])
                for attr in parts[1:]:
                    value = getattr(value, attr)
                d[k] = value
            results.append(cls(**d))
        return results

Demo:

import sys
from collections.abc import Coroutine
from typing import Annotated, Any, Callable

if sys.version_info >= (3, 11):
    from typing import Self
else:
    from typing_extensions import Self

from pydantic import BaseModel

from tortoise import Model, fields, run_async
from tortoise.contrib.test import init_memory_sqlite


class View(BaseModel):
    class Meta:
        model: type[Model]

    @classmethod
    async def filter(cls, **kw: Any) -> list[Self]:
        keys: dict[str, str] = {
            k: v.metadata and v.metadata[0] or k for k, v in cls.model_fields.items()
        }
        query: dict[str, Any] = {keys.get(k, k): v for k, v in kw.items()}
        qs = cls.Meta.model.filter(**query)
        field_parts: dict[str, list[str]] = {f: f.split("__") for f in keys.values()}
        if related := [ps[0] for ps in field_parts.values() if len(ps) > 1]:
            qs = qs.prefetch_related(*related)
        results: list[Self] = []
        async for obj in qs:
            d: dict[str, Any] = dict(keys)
            for k, v in keys.items():
                parts: list[str] = field_parts[v]
                value = getattr(obj, parts[0])
                for attr in parts[1:]:
                    value = getattr(value, attr)
                d[k] = value
            results.append(cls(**d))
        return results


class A(Model):
    name = fields.CharField(20)


class B(Model):
    name = fields.CharField(20)
    a: fields.ForeignKeyRelation[A] = fields.ForeignKeyField("models.A", on_delete=fields.CASCADE)


class C(View):
    b_id: Annotated[int, "id"]
    a_name: Annotated[str, "a__name"]

    class Meta:
        model = B


def run_in_memory_sqlite(func: Callable[[], Coroutine]) -> None:
    run_async(init_memory_sqlite(func)())


async def main() -> None:
    a = await A.create(name="a")
    b = await B.create(name="b", a=a)
    print(await C.filter(b_id=b.pk))
    c_list: list[C] = await C.filter(a_name=a.name)
    print(c_list[0].model_dump())


if __name__ == "__main__":
    run_in_memory_sqlite(main)

Output of the demo:

[C(b_id=1, a_name='a')]
{'b_id': 1, 'a_name': 'a'}

[merlinz01]

@waketzheng Interesting idea, but your solution is more like syntax sugar and doesn't provide the performance benefits of a true SQL view.

[merlinz01]

Note I have edited the OP with further specification about how it could/should work.