communex.module.module
Tools for defining Commune modules.
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):
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__
andModel.__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.
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:
@dataclass
class
EndpointDefinition22@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])
params_model: type[EndpointParams]
def
endpoint(fn: Callable[~P, ~T]) -> Callable[~P, ~T]:
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:
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
extract_endpoints(self):
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