mirror of
https://github.com/dbt-labs/dbt-core
synced 2025-12-17 19:31:34 +00:00
114 lines
2.8 KiB
Python
114 lines
2.8 KiB
Python
#!/usr/bin/env python
|
|
|
|
import inspect
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Iterable, List, Optional, Union
|
|
|
|
from dbt.context.base import BaseContext
|
|
from dbt.context.providers import MacroContext, ModelContext
|
|
from dbt.context.target import TargetContext
|
|
from dbt_common.dataclass_schema import dbtClassMixin
|
|
|
|
CONTEXTS_MAP = {
|
|
"base": BaseContext,
|
|
"target": TargetContext,
|
|
"model": ModelContext,
|
|
"macro": MacroContext,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ContextValue(dbtClassMixin):
|
|
name: str
|
|
value: str # a type description
|
|
doc: Optional[str]
|
|
|
|
|
|
@dataclass
|
|
class MethodArgument(dbtClassMixin):
|
|
name: str
|
|
value: str # a type description
|
|
|
|
|
|
@dataclass
|
|
class ContextMethod(dbtClassMixin):
|
|
name: str
|
|
args: List[MethodArgument]
|
|
result: str # a type description
|
|
doc: Optional[str]
|
|
|
|
|
|
@dataclass
|
|
class Unknown(dbtClassMixin):
|
|
name: str
|
|
value: str
|
|
doc: Optional[str]
|
|
|
|
|
|
ContextMember = Union[ContextValue, ContextMethod, Unknown]
|
|
|
|
|
|
def _get_args(func: inspect.Signature) -> Iterable[MethodArgument]:
|
|
found_first = False
|
|
for argname, arg in func.parameters.items():
|
|
if found_first is False and argname in {"self", "cls"}:
|
|
continue
|
|
if found_first is False:
|
|
found_first = True
|
|
|
|
yield MethodArgument(
|
|
name=argname,
|
|
value=inspect.formatannotation(arg.annotation),
|
|
)
|
|
|
|
|
|
def collect(cls):
|
|
values = []
|
|
for name, v in cls._context_members_.items():
|
|
attrname = cls._context_attrs_[name]
|
|
attrdef = getattr(cls, attrname)
|
|
doc = getattr(attrdef, "__doc__")
|
|
if inspect.isfunction(attrdef):
|
|
sig = inspect.signature(attrdef)
|
|
result = inspect.formatannotation(sig.return_annotation)
|
|
sig_good_part = ContextMethod(
|
|
name=name,
|
|
args=list(_get_args(sig)),
|
|
result=result,
|
|
doc=doc,
|
|
)
|
|
elif isinstance(attrdef, property):
|
|
sig = inspect.signature(attrdef.fget)
|
|
sig_txt = inspect.formatannotation(sig.return_annotation)
|
|
sig_good_part = ContextValue(name=name, value=sig_txt, doc=doc)
|
|
else:
|
|
sig_good_part = Unknown(name=name, value=repr(attrdef), doc=doc)
|
|
values.append(sig_good_part)
|
|
|
|
return values
|
|
|
|
|
|
@dataclass
|
|
class ContextCatalog(dbtClassMixin):
|
|
base: List[ContextMember]
|
|
target: List[ContextMember]
|
|
model: List[ContextMember]
|
|
macro: List[ContextMember]
|
|
schema: Dict[str, Any]
|
|
|
|
|
|
def main():
|
|
catalog = ContextCatalog(
|
|
base=collect(BaseContext),
|
|
target=collect(TargetContext),
|
|
model=collect(ModelContext),
|
|
macro=collect(MacroContext),
|
|
schema=ContextCatalog.json_schema(),
|
|
)
|
|
print(json.dumps(catalog.to_dict()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|