75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
from typing import Any
|
|
|
|
|
|
float_type = type(0.0).__name__
|
|
str_type = type('').__name__
|
|
server_point_type = type((0.0,0.0)).__name__
|
|
client_point_type = type({'x': 0.0, 'y': 0.0}).__name__
|
|
str_list_type = type(['']).__name__
|
|
|
|
|
|
def define_property(type: str, optional: bool = False, readonly: bool = False) -> dict[str, Any]:
|
|
return { 'type': type, 'optional': optional, 'readonly': readonly }
|
|
|
|
|
|
class Serialize(object):
|
|
def __init__(self, row, schema) -> None:
|
|
self.row = row
|
|
self.schema = schema
|
|
|
|
def to_execution(self, name) -> str:
|
|
value = self.row[name]
|
|
type = self.schema[name]['type']
|
|
|
|
if value == None:
|
|
return 'null'
|
|
|
|
if type == float_type:
|
|
return value
|
|
|
|
if type == str_type:
|
|
return f"'{value}'"
|
|
|
|
raise Exception(f"Fail to serialize {name} for execution!")
|
|
|
|
def to_storage(self, name) -> str:
|
|
value = self.row[name]
|
|
type = self.schema[name]['type']
|
|
|
|
if value == None:
|
|
return 'null'
|
|
|
|
if type == float_type:
|
|
return value
|
|
|
|
if type == str_type:
|
|
return f"''{value}''"
|
|
|
|
raise Exception(f"Fail to serialize {name} for storage!")
|
|
|
|
def to_execution(self):
|
|
row = self.row.copy()
|
|
|
|
for key, value in row.items():
|
|
if value == None:
|
|
row[key] = 'null'
|
|
elif self.schema[key]['type'] == str_type:
|
|
row[key] = f"'{row[key]}'"
|
|
elif self.schema[key]['type'] == client_point_type:
|
|
row[key] = f"'({value['x']},{value['y']})'"
|
|
|
|
return row
|
|
|
|
def to_storage(self):
|
|
row = self.row.copy()
|
|
|
|
for key, value in row.items():
|
|
if value == None:
|
|
row[key] = 'null'
|
|
elif self.schema[key] == str_type:
|
|
row[key] = f"''{row[key]}''"
|
|
elif self.schema[key]['type'] == client_point_type:
|
|
row[key] = f"''({value['x']},{value['y']})''"
|
|
|
|
return row
|