63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
from .sections import *
|
|
from .database import API_ADD, API_UPDATE, API_DELETE, ChangeSet, DbChangeSet, execute_command
|
|
from .s23_options_util import set_option_cmd, set_option_v3_cmd
|
|
from .del_cmd_raw import del_cascade_cmd
|
|
|
|
|
|
def add_cmd(name: str, cs: ChangeSet) -> DbChangeSet | None:
|
|
return None
|
|
|
|
|
|
def set_cmd(name: str, cs: ChangeSet) -> DbChangeSet | None:
|
|
type = cs.operations[0]['type']
|
|
|
|
if type == s23_option:
|
|
return set_option_cmd(name, cs)
|
|
elif type == s23_option_v3:
|
|
return set_option_v3_cmd(name, cs)
|
|
|
|
return None
|
|
|
|
|
|
def del_cmd(name: str, cs: ChangeSet) -> DbChangeSet | None:
|
|
return None
|
|
|
|
|
|
def execute_batch_command(name: str, cs: ChangeSet) -> ChangeSet:
|
|
css: list[DbChangeSet] = []
|
|
|
|
# for delete, generate cascade command
|
|
new_cs = ChangeSet()
|
|
for op in cs.operations:
|
|
if op['operation'] == API_DELETE:
|
|
new_cs.merge(del_cascade_cmd(name, ChangeSet(op)))
|
|
else:
|
|
new_cs.merge(ChangeSet(op))
|
|
|
|
try:
|
|
for op in new_cs.operations:
|
|
operation = op['operation']
|
|
|
|
r = None
|
|
|
|
if operation == API_ADD:
|
|
r = add_cmd(name, ChangeSet(op))
|
|
elif operation == API_UPDATE:
|
|
r = set_cmd(name, ChangeSet(op))
|
|
elif operation == API_DELETE:
|
|
r = del_cmd(name, ChangeSet(op))
|
|
|
|
if r == None:
|
|
print(f'ERROR: Build [{op}] returns None')
|
|
return ChangeSet()
|
|
|
|
css.append(r)
|
|
|
|
except:
|
|
return ChangeSet()
|
|
|
|
try:
|
|
return execute_command(name, DbChangeSet.from_list(css))
|
|
except:
|
|
return ChangeSet()
|