[proposal] Class Based Operations

[vitalik]

Read full proposal here - http://django-ninja.rest-framework.com/proposals/cbv/

To allow incapsulate common logic into class methods

from ninja import Router


router = Router()


@router.path('/project/{project_id}/tasks')
class Tasks:

    def __init__(self, request, project_id=int):
        user_projects = request.user.project_set
        self.project = get_object_or_404(user_projects, id=project_id))
        self.tasks = self.project.task_set.all()


    @router.get('/', response=List[TaskOut])
    def task_list(self, request):
        return self.tasks

    @router.get('/{task_id}/', response=TaskOut)
    def details(self, request, task_id: int):
        return get_object_or_404(self.tasks, id=task_id)

[rajaiswal]

@vitalik About the async def __init__(): issue, this seems like a good workaround:

import asyncio

dsn = "..."

class Foo(object):
    @classmethod
    async def create(cls, settings):
        self = Foo()
        self.settings = settings
        self.pool = await create_pool(dsn)
        return self

async def main(settings):
    settings = "..."
    foo = await Foo.create(settings)

Source: https://stackoverflow.com/questions/33128325/how-to-set-class-attribute-with-await-in-init/33134213
Couple other workarounds are also proposed in this post that we could look into.

[vitalik]

yes, this seems the only option... (without the classmethod), the onlyl thing I struggle how to name that method:

my shortlist so far

• init (without _ )
• create
• prepare
• before_request

[rajaiswal]

@vitalik First option init (without _ ) sounds good since that is what we're trying to replace.

[lqx131499]

I don't think we need this design. The design of NINJA is very good now. It can transplant the logic code to other frameworks, such as FASTAPI and FLASK.

[rhzs]

Hi @vitalik , why do we need to use __init__ ?

Can we use middleware and rails like approach?

from ninja import Router


router = Router()

@router.path('/project/{project_id}/tasks', before_action=['get_all_tasks'], after_action=['after_action'], around_action=['around_action'])
class Tasks:

    def get_all_tasks(self, request, project_id=int):
        user_projects = request.user.project_set
        self.project = get_object_or_404(user_projects, id=project_id))
        self.tasks = self.project.task_set.all()

    @router.get('/', response=List[TaskOut])
    def task_list(self, request):
        return self.tasks

    @router.get('/{task_id}/', response=TaskOut)
    def details(self, request, task_id: int):
        return get_object_or_404(self.tasks, id=task_id)

Or you may want to write like this..

@router.path('/project/{project_id}/tasks, around_action=['around_action'])
class Tasks:

    @router.before_action()
    def get_all_tasks(self, request, project_id=int):
        user_projects = request.user.project_set
        self.project = get_object_or_404(user_projects, id=project_id))
        self.tasks = self.project.task_set.all()

    @router.after_action()
    def log_action(self):
        print("hello")

    @router.get('/', response=List[TaskOut])
    def task_list(self, request):
        return self.tasks

    @router.get('/{task_id}/', response=TaskOut)
    def details(self, request, task_id: int):
        return get_object_or_404(self.tasks, id=task_id)

Ref: https://guides.rubyonrails.org/action_controller_overview.html

[FJLendinez]

Here the implementation of my proposal:

from functools import partial
from typing import List

from django.shortcuts import get_object_or_404
from ninja import Router


def update_object(obj, payload):
    for attr, value in payload.dict().items():
        if value:
            setattr(obj, attr, value)
    return obj


class ApiModel:
    RETRIEVE = 'retrieve'
    LIST = 'list'
    CREATE = 'create'
    UPDATE = 'update'
    DELETE = 'delete'
    views = [RETRIEVE, LIST, CREATE, UPDATE, DELETE]
    auth = None
    response_schema = None
    create_schema = None
    update_schema = None
    router: Router = None
    queryset = None

    def __init__(self):
        if not self.router:
            raise Exception("Router is necessary")
        if not self.response_schema:
            raise Exception("Response schema is necessary")

        if self.LIST in self.views:
            self.router.add_api_operation(path="/", methods=["GET"],
                                          auth=self.auth,
                                          view_func=self.get_list_func(),
                                          response=List[self.response_schema])
        if self.RETRIEVE in self.views:
            self.router.add_api_operation(path="/{pk}", methods=["GET"],
                                          auth=self.auth,
                                          view_func=self.get_retrieve_func(),
                                          response=self.response_schema)
        if self.UPDATE in self.views:
            self.router.add_api_operation(path="/{pk}", methods=["PUT", "PATCH"],
                                          auth=self.auth,
                                          view_func=self.get_put_path_func(),
                                          response=self.response_schema)
        if self.CREATE in self.views:
            self.router.add_api_operation(path="/", methods=["POST"],
                                          auth=self.auth,
                                          view_func=self.get_post_func(),
                                          response=self.response_schema)
        if self.DELETE in self.views:
            self.router.add_api_operation(path="/{pk}", methods=["DELETE"],
                                          auth=self.auth,
                                          view_func=self.get_delete_func())

    def get_list_func(self):
        model = self.queryset.model

        def list_func(get_queryset, request):
            return get_queryset(request)

        list_func = partial(list_func, self.get_queryset)
        list_func.__name__ = f"list_{model._meta.model_name}"
        return list_func

    def get_retrieve_func(self):
        model = self.queryset.model

        def retrieve_func(get_queryset, request, pk):
            return get_object_or_404(get_queryset(request), pk=pk)

        retrieve_func = partial(retrieve_func, self.get_queryset)
        retrieve_func.__name__ = f"retrieve_{model._meta.model_name}"
        return retrieve_func

    def get_post_func(self):
        create_schema = self.create_schema
        model = self.queryset.model

        def post_func(request, payload: create_schema):
            return self.queryset.model.objects.create(**payload.dict())

        post_func.__name__ = f"create_{model._meta.model_name}"
        return post_func

    def get_put_path_func(self):
        update_schema = self.update_schema
        model = self.queryset.model

        def put_path_func(request, pk, payload: update_schema):
            return update_object(
                get_object_or_404(self.get_queryset(request), pk=pk), payload)

        put_path_func.__name__ = f"update_{model._meta.model_name}"
        return put_path_func

    def get_delete_func(self):
        model = self.queryset.model

        def delete_func(request, pk):
            obj = get_object_or_404(self.get_queryset(request), pk=pk)
            obj.delete()
            return

        delete_func.__name__ = f"delete_{model._meta.model_name}"
        return delete_func

    def get_queryset(self, request):
        return self.queryset

Usage:

router = Router()

class CustomerAddressAPI(ApiModel):
    queryset = CustomerAddress.objects.all()
    auth = KnoxAuth() # https://github.com/FJLendinez/django-ninja-knox  
    create_schema = CreateCustomerAddressSchema
    update_schema = UpdateCustomerAddressSchema
    response_schema = OutCustomerAddressSchema
    router = router


# api.add_router('/customer-addresses/', CustomerAddressAPI().router)

[lorddaedra]

Just use functions...

[erhuabushuo]

It is a nice feature, When does it come out?

[Kojiro20]

I solved this in a pretty nice way for FastAPI but they just give me 👎 and no discussion @tiangolo please consider it 🦗: fastapi/fastapi#2626

My approach solves both issues - so you can actually use the class in multiple instances like this:

from fastapi import FastAPI
from fastapi.routing import ViewAPIRouter
from fastapi.testclient import TestClient
from pydantic import BaseModel

view_router = ViewAPIRouter()
app = FastAPI()


class Item(BaseModel):
    message_update: str


@view_router.bind_to_class()
class Messages(ViewAPIRouter.View):
    def __init__(self, message: str):
        self.message = message

    @view_router.get("/message")
    def get_message(self) -> str:
        return self.message

    @view_router.post("/message")
    def post_message(self, item: Item) -> str:
        self.message = item.message_update


em_instance = Messages(message="👋")
pt_instance = Messages(message="olá")
en_instance = Messages(message="hello")
app.include_router(em_instance.router, prefix="/em")
app.include_router(pt_instance.router, prefix="/pt")
app.include_router(en_instance.router, prefix="/en")
client = TestClient(app)


# verify route inclusion
response = client.get("/em/message")
assert response.status_code == 200, response.text
assert em_instance.message in response.text
response = client.get("/pt/message")
assert response.status_code == 200, response.text
assert pt_instance.message in response.text
response = client.get("/en/message")
assert response.status_code == 200, response.text
assert en_instance.message in response.text

# change state in an instance
item = Item(message_update="✨")
response = client.post("/em/message", json=item.dict())
assert response.status_code == 200, response.text
response = client.get("/em/message")
assert response.status_code == 200, response.text
assert item.message_update in response.text

The above code is functional in my fork and all the docs work as expected. The change is relatively simple - adds maybe 100 lines to encapsulate the class decorator and store the route decorations. Then it injects the 'self' parameter for each class instance on top of the provided values. It is a relatively small change with very high value.

@vitalik would you be interested in a similar PR to this project?

I think all the people who 👎 without much thought are missing the value of encapsulation and testability that this brings to the code. It's very pythonic to drop everything into global scope – but you lose the ability to instantiate fresh during tests, instead you need complex dependency mocks or tests without mocks. This does not scale in a large project with many contributors.

[vitalik]

Hi @Kojiro20

Well, what you do here is a bit different from what the goal is
you create instances on declaration time, while goal is that class instance created on each request, so that during instance initialisation some common attributes can be set to avoid code duplications

[Kojiro20]

I understand what you're trying to do - but if you'll bear with me a little - I'll try to explain what I meant by 'it solves both problems' above.

Django (and python at large) are very much biased toward whiz-bang trickery and auto-magic injection of things. This, still, despite guido leaving and everyone supposedly caring about the 'Principle of Least Astonishment'. The hot-take is that you wouldn't be trying to find a name for your post-init initializer if Django hadn't already created a paradigm where everything is side-loaded from config and confusingly-injected at runtime everywhere in the platform.

The only reason you need to decide between "init, create, prepare, before_request" is because Django has stolen the init function in its frenzied attempt to make WSGI scale. But, if you don't yield your init power to begin with, you won't need to reclaim it later.

This can be done if you init with resolve and re-use the instance you started with. Django actually allows this (which is why I'm not using FastAPI for my latest gig) but none of the niceties come along with it (so far). If you instantiated a class and made a "urls.py-concatatable" set of urls (and maybe even throw in some sane method semantics (lol, django has no sane method filtration semantics outside cbvs after 15yrs?!?) as a bonus) you could instantiate once and then concat something onto the urls list.

Sorry this is a bit of a rant. I just regret taking a contract that requires python. Since Guido left, everyone is breaking back-compat with wild abandon at the same time, while doubling down on the "Just use functions..." mentality. 30yrs in this industry and honestly the worst code that exists is written in c# (and derivatives (or perhaps superlatives?)) and python. Don't get me wrong, Java can be awful, but it's never as bad as others at their worst (nor as good when they're at their best).

[vitalik]

well still... looking at your example I do not see how can it solve the issue when:

you have a bunch of API urls which start with the same path/params

• GET /project/<proj-id>/tasks
• POST /project/<proj-id>/tasks
• GET /project/<proj-id>/tasks/<task-id>
• PATCH /project/<proj-id>/tasks/<task-id>
• DELETE /project/<proj-id>/tasks/<task-id>
• POST /project/<proj-id>/tasks/<task-id>/complete

as you can see a little todo app - there are 6 methods that share the same logic for checking permission (if user have access to project)

user_projects = request.user.project_set
self.project = get_object_or_404(user_projects, id=project_id))
self.tasks = self.project.task_set.all()

repeating these in every operation will lead to tons of code duplications...