Edit on GitHub

communex.module.module

Tools for defining Commune modules.

View Source

 1"""
 2Tools for defining Commune modules.
 3"""
 4
 5import inspect
 6from dataclasses import dataclass
 7from typing import Any, Callable, Generic, ParamSpec, TypeVar, cast
 8
 9import pydantic
10from pydantic import BaseModel
11
12T = TypeVar('T')
13P = ParamSpec('P')
14
15
16class EndpointParams(BaseModel):
17    class config:
18        extra = "allow"
19
20
21@dataclass
22class EndpointDefinition(Generic[T, P]):
23    name: str
24    fn: Callable[P, T]
25    params_model: type[EndpointParams]
26
27
28def endpoint(fn: Callable[P, T]) -> Callable[P, T]:
29    sig = inspect.signature(fn)
30    params_model = function_params_to_model(sig)
31    name = fn.__name__
32
33    endpoint_def = EndpointDefinition(name, fn, params_model)
34    fn._endpoint_def = endpoint_def  # type: ignore
35
36    return fn
37
38
39def function_params_to_model(signature: inspect.Signature) -> type[EndpointParams]:
40    fields: dict[str, tuple[type] | tuple[type, Any]] = {}
41    for i, param in enumerate(signature.parameters.values()):
42        name = param.name
43        if name == "self":  # cursed
44            assert i == 0
45            continue
46        annotation = param.annotation
47        if annotation == param.empty:
48            raise Exception(
49                f"Error: annotation for parameter `{name}` not found")
50
51        if param.default == param.empty:
52            fields[name] = (annotation, ...)
53        else:
54            fields[name] = (annotation, param.default)
55
56    model: type[EndpointParams] = cast(
57
58        type[EndpointParams], pydantic.create_model(  #  type: ignore
59            'Params', **fields, __base__=EndpointParams)  #  type: ignore
60    )
61
62    return model
63
64
65class Module:
66    def __init__(self) -> None:
67        # TODO: is it possible to get this at class creation instead of object instantiation?
68        self.__endpoints = self.extract_endpoints()
69
70    def get_endpoints(self):
71        return self.__endpoints
72
73    def extract_endpoints(self):
74        endpoints: dict[str, EndpointDefinition[Any, Any]] = {}
75        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
76            if hasattr(method, '_endpoint_def'):
77                endpoint_def: EndpointDefinition = method._endpoint_def  # type: ignore
78                endpoints[name] = endpoint_def  # type: ignore
79        return endpoints

P = ~P

class EndpointParams(pydantic.main.BaseModel): View Source

17class EndpointParams(BaseModel):
18    class config:
19        extra = "allow"

Usage docs: https://docs.pydantic.dev/2.7/concepts/models/

A base class for creating Pydantic models.

Attributes:

__class_vars__: The names of classvars defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The signature for instantiating the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The pydantic-core schema used to build the SchemaValidator and SchemaSerializer.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to __args__, __origin__, __parameters__ in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a RootModel.
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_extra__: An instance attribute with the values of extra fields from validation when model_config['extra'] == 'allow'.
__pydantic_fields_set__: An instance attribute with the names of fields explicitly set.
__pydantic_private__: Instance attribute with the values of private attributes set on the model instance.

model_config = {}

model_fields = {}

model_computed_fields = {}

Inherited Members

pydantic.main.BaseModel: BaseModel; model_extra; model_fields_set; model_construct; model_copy; model_dump; model_dump_json; model_json_schema; model_parametrized_name; model_post_init; model_rebuild; model_validate; model_validate_json; model_validate_strings; dict; json; parse_obj; parse_raw; parse_file; from_orm; construct; copy; schema; schema_json; validate; update_forward_refs

class EndpointParams.config: View Source

18    class config:
19        extra = "allow"

extra = 'allow'

@dataclass

class EndpointDefinition(typing.Generic[~T, ~P]): View Source

22@dataclass
23class EndpointDefinition(Generic[T, P]):
24    name: str
25    fn: Callable[P, T]
26    params_model: type[EndpointParams]

EndpointDefinition( name: str, fn: Callable[~P, ~T], params_model: type[EndpointParams])

name: str

fn: Callable[~P, ~T]

params_model: type[EndpointParams]

def endpoint(fn: Callable[~P, ~T]) -> Callable[~P, ~T]: View Source

29def endpoint(fn: Callable[P, T]) -> Callable[P, T]:
30    sig = inspect.signature(fn)
31    params_model = function_params_to_model(sig)
32    name = fn.__name__
33
34    endpoint_def = EndpointDefinition(name, fn, params_model)
35    fn._endpoint_def = endpoint_def  # type: ignore
36
37    return fn

def function_params_to_model( signature: inspect.Signature) -> type[EndpointParams]: View Source

40def function_params_to_model(signature: inspect.Signature) -> type[EndpointParams]:
41    fields: dict[str, tuple[type] | tuple[type, Any]] = {}
42    for i, param in enumerate(signature.parameters.values()):
43        name = param.name
44        if name == "self":  # cursed
45            assert i == 0
46            continue
47        annotation = param.annotation
48        if annotation == param.empty:
49            raise Exception(
50                f"Error: annotation for parameter `{name}` not found")
51
52        if param.default == param.empty:
53            fields[name] = (annotation, ...)
54        else:
55            fields[name] = (annotation, param.default)
56
57    model: type[EndpointParams] = cast(
58
59        type[EndpointParams], pydantic.create_model(  #  type: ignore
60            'Params', **fields, __base__=EndpointParams)  #  type: ignore
61    )
62
63    return model

class Module: View Source

66class Module:
67    def __init__(self) -> None:
68        # TODO: is it possible to get this at class creation instead of object instantiation?
69        self.__endpoints = self.extract_endpoints()
70
71    def get_endpoints(self):
72        return self.__endpoints
73
74    def extract_endpoints(self):
75        endpoints: dict[str, EndpointDefinition[Any, Any]] = {}
76        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
77            if hasattr(method, '_endpoint_def'):
78                endpoint_def: EndpointDefinition = method._endpoint_def  # type: ignore
79                endpoints[name] = endpoint_def  # type: ignore
80        return endpoints

def get_endpoints(self): View Source

71    def get_endpoints(self):
72        return self.__endpoints

def extract_endpoints(self): View Source

74    def extract_endpoints(self):
75        endpoints: dict[str, EndpointDefinition[Any, Any]] = {}
76        for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
77            if hasattr(method, '_endpoint_def'):
78                endpoint_def: EndpointDefinition = method._endpoint_def  # type: ignore
79                endpoints[name] = endpoint_def  # type: ignore
80        return endpoints