tweak: Use Generic to support 3.11

This commit is contained in:
2026-05-31 06:31:46 +10:00
parent 4734265f2e
commit 733f3e3ae6
+46 -30
View File
@@ -1,4 +1,14 @@
from typing import Any, Awaitable, Callable, Concatenate, ParamSpec, TypeVar, Union, TYPE_CHECKING
from typing import (
Any,
Generic,
Awaitable,
Callable,
Concatenate,
ParamSpec,
TypeVar,
Union,
TYPE_CHECKING,
)
from functools import wraps, reduce
import operator
@@ -22,9 +32,9 @@ __all__ = (
"ternary_term",
"run_function",
"cascade_term",
'ExecutorTerm',
'LateTerm',
'lazy_term',
"ExecutorTerm",
"LateTerm",
"lazy_term",
)
P = ParamSpec("P")
@@ -37,7 +47,7 @@ T2 = TypeVar("T2")
TermResult = TypeVar("TermResult")
class ExecutorTerm[TermResult]:
class ExecutorTerm(Generic[TermResult]):
# May add more here for initial parsing context
def __init__(self, *args, node_text: str | None = None, **kwargs):
@@ -145,7 +155,7 @@ async def cmdarg_term(ctx, argn: int, greedy: bool) -> str:
if greedy:
return " ".join(args[argn:])
elif argn >= len(args):
return ''
return ""
else:
return args[argn]
@@ -158,33 +168,36 @@ async def cascade_term(ctx, *terms: ExecutorTerm):
return eval
return ""
@lazy_term
async def command_shell(ctx, component_term: ExecutorTerm):
if isinstance(component_term, ExecutorTerm):
result = await component_term(ctx)
else:
logger.warning(
f"command_shell LazyTerm stringifying non-term {component_term}"
)
logger.warning(f"command_shell LazyTerm stringifying non-term {component_term}")
result = component_term
return str(result)
@lazy_term
async def define_equal(ctx, var: str, value: ExecutorTerm):
ctx.this_scope[var] = await value.execute(ctx)
return None
@lazy_term
async def define_walrus(ctx, var: str, value: ExecutorTerm):
result = await value.execute(ctx)
ctx.this_scope[var] = result
return result
@lazy_term
@lazy_term
async def define_term(ctx, var: str, value: ExecutorTerm):
ctx.this_scope[var] = value
return None
@lazy_term
async def run_function(ctx, var: ExecutorTerm, *args: ExecutorTerm):
varlue = await var.execute(ctx)
@@ -196,13 +209,14 @@ async def run_function(ctx, var: ExecutorTerm, *args: ExecutorTerm):
else:
return await varlue(ctx, *argues)
@lazy_term
@lazy_term
async def get_var(ctx, root: str | ExecutorTerm, *attrs: str):
if isinstance(root, ExecutorTerm):
value = await root.execute(ctx)
name = [root.text or '']
name = [root.text or ""]
elif root not in ctx.scope:
return ' '.join((root, *attrs))
return " ".join((root, *attrs))
else:
value = ctx.scope[root]
name = [root]
@@ -213,39 +227,41 @@ async def get_var(ctx, root: str | ExecutorTerm, *attrs: str):
for attr in attrs:
try:
value = getattr(value, '_exec_attr_'+attr)
value = getattr(value, "_exec_attr_" + attr)
name.append(attr)
except AttributeError:
raise CommandValueError(f"{'.'.join(name)} has no attribute {attr}")
return value
@lazy_term
@lazy_term
async def compare(ctx, term1: ExecutorTerm, op: str, term2: ExecutorTerm):
opmap = {
'==': operator.eq,
'!=': operator.ne,
'>': operator.gt,
'>=': operator.ge,
'<': operator.lt,
'<=': operator.le,
"==": operator.eq,
"!=": operator.ne,
">": operator.gt,
">=": operator.ge,
"<": operator.lt,
"<=": operator.le,
}
value1 = await term1.execute(ctx)
value2 = await term2.execute(ctx)
oppy = opmap[op]
return oppy(value1, value2)
@lazy_term
async def binop(ctx, op: str, *terms):
opmap = {
'||': operator.or_,
'&&': operator.and_,
'+': operator.add,
'-': operator.sub,
'*': operator.mul,
'**': operator.pow,
'/': operator.truediv,
'//': operator.floordiv,
'%': operator.mod,
"||": operator.or_,
"&&": operator.and_,
"+": operator.add,
"-": operator.sub,
"*": operator.mul,
"**": operator.pow,
"/": operator.truediv,
"//": operator.floordiv,
"%": operator.mod,
}
# Left associative
oppy = opmap[op]