Spaces:
Running
Running
File size: 1,273 Bytes
ca01fa3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
'''API for implementing LynxKite operations.'''
import dataclasses
import inspect
ALL_OPS = {}
@dataclasses.dataclass
class Op:
func: callable
name: str
params: dict
inputs: dict
outputs: dict
type: str
def __call__(self, *inputs, **params):
# Do conversions here.
res = self.func(*inputs, **params)
return res
@dataclasses.dataclass
class EdgeDefinition:
df: str
source_column: str
target_column: str
source_table: str
target_table: str
source_key: str
target_key: str
@dataclasses.dataclass
class Bundle:
dfs: dict
edges: list[EdgeDefinition]
def op(name):
'''Decorator for defining an operation.'''
def decorator(func):
type = func.__annotations__.get('return') or 'basic'
sig = inspect.signature(func)
# Positional arguments are inputs.
inputs = {
name: param.annotation
for name, param in sig.parameters.items()
if param.kind != param.KEYWORD_ONLY}
params = {
name: param.default if param.default is not inspect._empty else None
for name, param in sig.parameters.items()
if param.kind == param.KEYWORD_ONLY}
op = Op(func, name, params=params, inputs=inputs, outputs={'output': 'yes'}, type=type)
ALL_OPS[name] = op
return func
return decorator
|