Descriptors

gistfile1.txt

import asyncio
import dataclasses
from asyncio import iscoroutine
from datetime import datetime
from typing import Union, Optional

from pydantic import BaseModel, Field, validator, validate_arguments, UUID4
import uuid


class CompleteAsyncDescriptors:
    async def complete(self):
        coroutines = [(attr, val) for attr, val in self.__dict__.items() if iscoroutine(val)]
        if coroutines:
            results = await asyncio.gather(*(c[-1] for c in coroutines))
            for i, (attr, _) in enumerate(coroutines):
                self.__dict__[attr] = results[i]
        return self


class field:
    # init that recieves the setter and getter
    def __init__(self, getter=None, setter=None):
        self.getter = getter
        self.setter = setter

    def __set_name__(self, owner, name):
        self.name = name
        self.setter = self.setter or f"set_{self.name}"
        self.getter = self.getter or f"get_{self.name}"

    def __set__(self, instance, value):
        if setter_meth := getattr(instance, self.setter, None):
            value = setter_meth(value)

        instance.__dict__[self.name] = value
        return getattr(instance, self.name)


async def get_user_by_id_async(uid: str):
    print(f"get_user_by_id_async {uid}")
    return f"Вася Пупкин "


@dataclasses.dataclass
class SubProcessAsync(CompleteAsyncDescriptors):
    blabla: str
    initiator: Optional[str] = field(setter="huy")
    executor: Optional[str] = field()

    async def set_initiator(self, value: str) -> str:
        return await get_user_by_id_async(value)

    async def set_executor(self, value: str) -> str:
        return await get_user_by_id_async(value)


if __name__ == '__main__':
    asyncio.run(SubProcessAsync(blabla="blabla", initiator="124", executor="123").complete())