Session

Session Interface

class gs_interactive.client.session.Session

DefaultSession Class

class gs_interactive.client.session.DefaultSession(admin_uri: str, stored_proc_uri: str | None = None)

The default session implementation for Interactive SDK. It provides the implementation of all service APIs.

__init__(admin_uri: str, stored_proc_uri: str | None = None)

Construct a new session using the specified admin_uri and stored_proc_uri.

Parameters:

• admin_uri (str) – the uri for the admin service.

• stored_proc_uri (str, optional) – the uri for the stored procedure service. If not provided,the uri will be read from the service status.

add_edge(graph_id: Annotated[str, Strict(strict=True)], edge_request: List[EdgeRequest]) → Result[str]

Add an edge to the specified graph.

add_vertex(graph_id: Annotated[str, Strict(strict=True)], vertex_edge_request: VertexEdgeRequest) → Result[Annotated[str, Strict(strict=True)]]

Add a vertex to the specified graph.

bulk_loading(graph_id: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The id of graph to load')], schema_mapping: SchemaMapping) → Result[JobResponse]

Submit a bulk loading job to the specified graph.

call_procedure(graph_id: Annotated[str, Strict(strict=True)], params: QueryRequest) → Result[CollectiveResults]

Call a procedure in the specified graph.

call_procedure_current(params: QueryRequest) → Result[CollectiveResults]

Call a procedure in the current graph.

call_procedure_current_raw(params: bytes) → Result[str]

Call a procedure in the current graph with raw bytes.

call_procedure_raw(graph_id: Annotated[str, Strict(strict=True)], params: bytes) → Result[str]

Call a procedure in the specified graph with raw bytes.

cancel_job(job_id: Annotated[str, Strict(strict=True)]) → Result[str]

Cancel a job with the specified job id.

create_edge_type(graph_id: Annotated[str, Strict(strict=True)], create_edge_type: CreateEdgeType) → Result[str]

Create a new edge type in the specified graph.

create_graph(graph: CreateGraphRequest) → Result[CreateGraphResponse]

Create a new graph with the specified graph request.

create_procedure(graph_id: Annotated[str, Strict(strict=True)], procedure: CreateProcedureRequest) → Result[CreateProcedureResponse]

Create a new procedure in the specified graph.

create_vertex_type(graph_id: Annotated[str, Strict(strict=True)], create_vertex_type: CreateVertexType) → Result[str]

Create a new vertex type in the specified graph.

delete_edge(graph_id: Annotated[str, Strict(strict=True)], delete_edge_request: Annotated[List[DeleteEdgeRequest], FieldInfo(annotation=NoneType, required=True, description='The label and primary key values of the src and dst vertices, and the edge label.')]) → Result[str]

Delete an edge from the specified graph with primary key value.

delete_edge_type(graph_id: Annotated[str, Strict(strict=True)], type_name: Annotated[str, Strict(strict=True)], source_vertex_type: Annotated[str, Strict(strict=True)], destination_vertex_type: Annotated[str, Strict(strict=True)]) → Result[str]

Delete a edge type in the specified graph.

delete_graph(graph_id: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The id of graph to delete')]) → Result[str]

Delete a graph with the specified graph id.

delete_procedure(graph_id: Annotated[str, Strict(strict=True)], procedure_id: Annotated[str, Strict(strict=True)]) → Result[str]

Delete a procedure in the specified graph.

delete_vertex(graph_id: Annotated[str, Strict(strict=True)], delete_vertex_request: Annotated[List[DeleteVertexRequest], FieldInfo(annotation=NoneType, required=True, description='The label and primary key values of the vertex to be deleted.')]) → Result[str]

Delete a vertex from the specified graph with primary key value.

delete_vertex_type(graph_id: Annotated[str, Strict(strict=True)], type_name: Annotated[str, Strict(strict=True)]) → Result[str]

Delete a vertex type in the specified graph.

ensure_param_str(param_name: str, param)

Ensure the param is a string, otherwise raise an exception

get_edge(graph_id: Annotated[str, Strict(strict=True)], edge_label: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The label name of edge.')], src_label: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The label name of src vertex.')], src_primary_key_value: Annotated[Any, FieldInfo(annotation=NoneType, required=True, description='The primary key value of src vertex.')], dst_label: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The label name of dst vertex.')], dst_primary_key_value: Annotated[Any, FieldInfo(annotation=NoneType, required=True, description='The primary key value of dst vertex.')]) → Result[None | EdgeRequest]

Get an edge from the specified graph with primary key value.

get_graph_meta(graph_id: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The id of graph to get')]) → Result[GetGraphResponse]

Get the meta information of a specified graph.

get_graph_schema(graph_id: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The id of graph to get')]) → Result[GetGraphSchemaResponse]

Get the schema of a specified graph.

Parameters:

graph_id (str) – The ID of the graph whose schema is to be retrieved.

Returns:

The result containing the schema of

the specified graph.

Return type:

Result[GetGraphSchemaResponse]

get_graph_statistics(graph_id: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The id of graph to get')]) → Result[GetGraphStatisticsResponse]

Get the statistics of a specified graph.

get_job(job_id: Annotated[str, Strict(strict=True)]) → Result[JobStatus]

Get the status of a job with the specified job id.

get_procedure(graph_id: Annotated[str, Strict(strict=True)], procedure_id: Annotated[str, Strict(strict=True)]) → Result[GetProcedureResponse]

Get a procedure in the specified graph.

get_service_status() → Result[ServiceStatus]

Get the status of the service.

get_snapshot_status(graph_id: Annotated[str, Strict(strict=True)], snapshot_id: Annotated[int, Strict(strict=True)]) → SnapshotStatus

Get the status of a snapshot with the specified snapshot id.

get_vertex(graph_id: Annotated[str, Strict(strict=True)], label: Annotated[str, Strict(strict=True), FieldInfo(annotation=NoneType, required=True, description='The label name of vertex.')], primary_key_value: Annotated[Any, FieldInfo(annotation=NoneType, required=True, description='The primary key value of vertex.')]) → Result[VertexData]

Get a vertex from the specified graph with primary key value.

list_graphs() → Result[List[GetGraphResponse]]

List all graphs.

list_jobs() → Result[List[JobResponse]]

List all jobs.

list_procedures(graph_id: Annotated[str, Strict(strict=True)]) → Result[List[GetProcedureResponse]]

List all procedures in the specified graph.

restart_service() → Result[str]

Restart the service.

start_service(start_service_request: Annotated[StartServiceRequest | None, FieldInfo(annotation=NoneType, required=True, description='Start service on a specified graph')] | None = None) → Result[str]

Start the service on a specified graph.

stop_service(graph_id: str | None = None) → Result[str]

Stop the service.

try_upload_files(schema_mapping: SchemaMapping) → Result[SchemaMapping]

Try to upload the input files if they are specified with a starting @ for input files in schema_mapping. Replace the path to the uploaded file with the path returned from the server.

The @ can be added to the beginning of data_source.location in schema_mapping.loading_config,or added to each file in vertex_mappings and edge_mappings.

1. location: @/path/to/dir

   inputs:

   • @/path/to/file1

   • @/path/to/file2

2. location: /path/to/dir

   inputs:

   • @/path/to/file1

   • @/path/to/file2

3. location: @/path/to/dir

   inputs:

   • /path/to/file1

   • /path/to/file2

4. location: /path/to/dir

   inputs:

   • /path/to/file1

   • /path/to/file2

4. location: None

   inputs:

   • @/path/to/file1

   • @/path/to/file2

Among the above 4 cases, only the 1, 3, 5 case are valid, for 2,4 the file will not be uploaded

update_edge(graph_id: Annotated[str, Strict(strict=True)], edge_request: List[EdgeRequest]) → Result[str]

Update an edge in the specified graph.

update_edge_type(graph_id: Annotated[str, Strict(strict=True)], update_edge_type: CreateEdgeType) → Result[str]

Update a edge type in the specified graph.

update_procedure(graph_id: Annotated[str, Strict(strict=True)], proc_id: Annotated[str, Strict(strict=True)], procedure: UpdateProcedureRequest) → Result[str]

Update a procedure in the specified graph.

update_vertex(graph_id: Annotated[str, Strict(strict=True)], vertex_request: VertexEdgeRequest) → Result[str]

Update a vertex in the specified graph.

update_vertex_type(graph_id: Annotated[str, Strict(strict=True)], update_vertex_type: CreateVertexType) → Result[str]

Update a vertex type in the specified graph.

upload_and_replace_input_inplace(schema_mapping: SchemaMapping) → Result[SchemaMapping]

For each input file in schema_mapping, if the file starts with @, upload the file to the server, and replace the path with the path returned from the server.

upload_file(filestorage: Annotated[bytes, Strict(strict=True)] | Annotated[str, Strict(strict=True)] | None) → Result[UploadFileResponse]

Upload a file to the server.