from typing import List, Dict
from ultipa.printer.prettyPrint import PrettyPrint, PrettyTable
from ultipa.structs.Graph import Graph
from ultipa.structs.Property import Property
from ultipa.types.types import BaseModel, Node, Edge, DataItem, Status, \
UltipaStatistics, ReturnReq, BaseUqlReply, ResultType, ResultAlias, ExplainPlan
from ultipa.utils.convert import Any
[docs]
class PropertyTable(BaseModel):
name: str
data: List[Property]
def __init__(self, name, data):
self.name = name
self.data = data
[docs]
class Response(BaseModel):
def __init__(self, status: Status = None, data: BaseUqlReply = None,
req: ReturnReq = None, statistics: UltipaStatistics = None, aliases: List[ResultAlias] = None):
self.status = status
self.data = data
self.statistics = statistics
self.req = req
self.aliases = aliases
[docs]
def Print(self):
pretty = PrettyPrint()
pretty.prettyStatus(self.status)
if self.status.code != 0:
return
if self.statistics:
pretty.prettyStatistics(self.statistics)
if isinstance(self.data, list):
dict_list = []
for i in self.data:
if isinstance(i,PropertyTable):
dict_list.append(i.toDict())
else:
dict_list.append(i.__dict__)
pretty.prettyDataList(dict_list)
if isinstance(self.data, Any):
dict_list = self.data.__dict__
pretty.prettyData(dict_list)
[docs]
class UltipaResponse(BaseModel):
def __init__(self, status: Status = None, items: Dict = None,
req: ReturnReq = None, statistics: UltipaStatistics = None, aliases: List[ResultAlias] = None,
explainPlan: List[ExplainPlan] = None):
self.status = status
self.items = items
self.aliases = aliases
self.req = req
self.statistics = statistics
self.explainPlan = explainPlan
[docs]
def alias(self, alias: str) -> DataItem:
if self.items == None:
return DataItem(alias, None, ResultType.getTypeStr(ResultType.RESULT_TYPE_UNSET))
if self.items.get(alias):
return self.items.get(alias)
return DataItem(alias, None, ResultType.getTypeStr(ResultType.RESULT_TYPE_UNSET))
[docs]
def get(self, index: int) -> DataItem:
if len(self.aliases) - 1 >= index:
data = self.items.get(self.aliases[index].alias)
if data:
return data
if self.aliases[index].result_type == ResultType.getTypeStr(ResultType.RESULT_TYPE_UNSET):
return DataItem(None, None, ResultType.getTypeStr(ResultType.RESULT_TYPE_UNSET))
else:
return data
return DataItem(None, None, ResultType.getTypeStr(ResultType.RESULT_TYPE_UNSET))
[docs]
def getExplainPlan(self):
return self.explainPlan
[docs]
def Print(self):
pretty = PrettyPrint()
pretty.prettyStatus(self.status)
if self.status.code != 0:
return
pretty.prettyStatistics(self.statistics)
explains = []
for key in self.items:
dataItem = self.items.get(key)
if dataItem.type == ResultType.getTypeStr(ResultType.RESULT_TYPE_NODE):
pretty.prettyNode(dataItem)
elif dataItem.type == ResultType.getTypeStr(ResultType.RESULT_TYPE_EDGE):
pretty.prettyEdge(dataItem)
elif dataItem.type == ResultType.getTypeStr(ResultType.RESULT_TYPE_TABLE):
pretty.prettyTable(dataItem)
elif dataItem.type == ResultType.getTypeStr(ResultType.RESULT_TYPE_PATH):
pretty.prettyPath(dataItem)
elif dataItem.type == ResultType.getTypeStr(ResultType.RESULT_TYPE_ATTR):
pretty.prettyAttr(dataItem)
if self.explainPlan:
for explan in self.explainPlan:
explains.append(explan)
if explains:
pretty.prettyTree(explains)
[docs]
class ResponseCommon(Response):
data: None
[docs]
class User(BaseModel):
username: str
create: str
last_login_time: str
graphPrivileges: dict
systemPrivileges: List[str]
policies: List[str]
[docs]
class Index(BaseModel):
name: str
properties: str
schema: str
status: str
[docs]
class IndexTable:
name: str
data: List[Index]
[docs]
class Stat(BaseModel):
cpuUsage: str
memUsage: str
company: str
cpuCores: str
expiredDate: str
serverType: str
version: str
[docs]
class Privilege(BaseModel):
graphPrivileges: List[str]
systemPrivileges: List[str]
[docs]
class Policy(BaseModel):
name: str
policies: List[str]
graphPrivileges: dict
systemPrivileges: List[str]
[docs]
class Top(BaseModel):
process_id: str
process_uql: str
duration: str
[docs]
class Return_Type(BaseModel):
is_realtime: bool
is_visualization: bool
is_wirte_back: bool
[docs]
class Task_info(BaseModel):
task_id: int
algo_name: str
start_time: int
writing_start_time: int
end_time: int
time_cost: int
TASK_STATUS: int
status_code: str
engine_cost: int
return_type: Return_Type
[docs]
class Task(BaseModel):
param: dict
task_info: Task_info
result: dict
[docs]
class SearchKhop(BaseModel):
values: dict
nodes: List[Node]
[docs]
class Path(BaseModel):
nodes: List[Node]
edges: List[Edge]
[docs]
class SearchPath(BaseModel):
paths: List[Path]
[docs]
class NodeSpread(SearchPath):
pass
[docs]
class AutoNet(SearchPath):
pass
[docs]
class AlgoResultOpt(BaseModel):
can_realtime: bool
can_visualization: bool
can_write_back: bool
[docs]
class Algo(BaseModel):
param: dict
name: str
result_opt: AlgoResultOpt
[docs]
class ResponseGraph(Response):
data: Graph
[docs]
class ResponseListGraph(Response):
data: List[Graph]
[docs]
class ResponseListIndex(UltipaResponse):
data: List[IndexTable]
[docs]
class ResponseListFulltextIndex(ResponseListIndex):
pass
[docs]
class ResponseSearchEdge(UltipaResponse):
items: List[DataItem]
[docs]
class ResponseSearchNode(UltipaResponse):
items: List[DataItem]
[docs]
class ResponseBulk:
uuids: List[int]
errorItem: Dict
[docs]
class ResponseInsert(Response):
data: List[int]
[docs]
class ResponseBatchAutoInsert(Response):
data: Dict[str, ResponseBulk]
[docs]
class InsertResponse(Response):
data: ResponseBulk
[docs]
class ResponseListPolicy(Response):
data: List[Policy]
[docs]
class ResponsePolicy(Response):
data: Policy
[docs]
class ResponseListPrivilege(Response):
data: List[Privilege]
[docs]
class ResponseListProperty(Response):
data: List[PropertyTable]
[docs]
class ResponseGetProperty(Response):
data: List[Property]
[docs]
class ResponseListTop(Response):
data: List[Top]
[docs]
class ResponseListTask(Response):
data: List[Task]
[docs]
class ResponseUser(Response):
data: User
[docs]
class ResponseListUser(Response):
data: List[User]
[docs]
class ResponseListAlgo(Response):
data: List[Algo]
[docs]
class Cluster:
host: str
status: bool
cpuUsage: str
memUsage: str
isLeader: bool
isFollowerReadable: bool
isAlgoExecutable: bool
isUnset: bool
[docs]
class ClusterInfo(Response):
data: List[Cluster]
[docs]
class ResponseStat(Response):
data: Stat
[docs]
class ResponseExport(Response):
data: List[Node]