Use a driver for application code. The Prism API defines the messages and transport conventions needed to implement a client. Use the Prism API release that matches your server and consult the protocol reference for message fields.
Obtain the bindings
The Prism API 2.2 release includes the Protobuf schemas and a Go archive. The definitions are also published as org.polypheny:prism:2.2 on Maven Central, polypheny-prism-api==2.2 on PyPI, and Polypheny.Prism version 2.2.0 on NuGet. The Maven JAR contains schemas for code generation. For other languages, generate bindings from the schemas with protoc.
To generate Python bindings yourself, unpack the release’s protobuf-files-2.2.zip and run the following in the directory containing org. The compiler and runtime versions in this example support the optional fields in the definitions.
python -m pip install grpcio-tools==1.64.1 protobuf==5.27.2
python -m grpc_tools.protoc -I . --python_out=. org/polypheny/prism/*.proto
Transport and session
- Open a TCP connection to the Prism port, or connect to the configured Unix domain socket. Both peers send the transport version prefixed by a single byte giving the length, including the final newline. The transport version strings are
plain-v1@polypheny.comfor TCP andunix-v1@polypheny.comfor Unix sockets. Verify the peer sends the same string. - After the transport handshake, prefix every serialized Protobuf message with its byte length as an eight-byte little-endian integer. Read the full prefix and payload, handling partial socket reads.
- Send a
Requestwith a positive ID and aconnection_requestas the first message. Supply the Prism API version, credentials, and any connection properties. CheckConnectionResponse.is_compatiblebefore sending queries. TCP authentication requires a username and password. With a Unix socket, the server can use the peer’s operating system credentials when the client omits the username. - Match each
Responseto its request by ID. Read responses untilResponse.lastis true. Handle anerror_responseas an error for that request. For query results,Frame.is_lastseparately tells you when to stop sendingFetchRequestmessages. - Close statement handles when finished. When auto-commit is disabled, use the commit and rollback requests to manage transactions. End the session with
DisconnectRequestand close the socket.
Run a query over TCP
This example requires Python 3.9 or later and a running Polypheny server with Prism 2.2. It uses 127.0.0.1:20590, the local pa user, and an empty password by default. Set PRISM_HOST, PRISM_PORT, PRISM_USER, and PRISM_PASSWORD as needed. Install the bindings in a Python virtual environment.
python -m pip install polypheny-prism-api==2.2
Save as prism_query.py and run it with Python. The example fetches two rows one at a time and prints Ada and Lin.
import os
import socket
import struct
from itertools import count
from org.polypheny.prism import protointerface_pb2 as protocol
def read_exact(sock, size):
data = bytearray()
while len(data) < size:
chunk = sock.recv(size - len(data))
if not chunk:
raise EOFError("Prism closed the connection")
data.extend(chunk)
return bytes(data)
def exchange(sock, request):
payload = request.SerializeToString()
sock.sendall(struct.pack("<Q", len(payload)) + payload)
while True:
(size,) = struct.unpack("<Q", read_exact(sock, 8))
response = protocol.Response.FromString(read_exact(sock, size))
if response.id != request.id:
raise RuntimeError(f"Unexpected response ID {response.id}")
if response.HasField("error_response"):
raise RuntimeError(response.error_response.message)
if response.last:
return response
address = (os.getenv("PRISM_HOST", "127.0.0.1"), int(os.getenv("PRISM_PORT", "20590")))
request_ids = count(1)
with socket.create_connection(address, timeout=10) as sock:
version = b"plain-v1@polypheny.com\n"
handshake = bytes([len(version)]) + version
sock.sendall(handshake)
(size,) = read_exact(sock, 1)
if read_exact(sock, size) != version:
raise RuntimeError("Unexpected Prism transport version")
request = protocol.Request(id=next(request_ids))
request.connection_request.major_api_version = 2
request.connection_request.minor_api_version = 2
request.connection_request.username = os.getenv("PRISM_USER", "pa")
request.connection_request.password = os.getenv("PRISM_PASSWORD", "")
response = exchange(sock, request).connection_response
if not response.is_compatible:
server_api = f"{response.major_api_version}.{response.minor_api_version}"
raise RuntimeError(f"Incompatible Prism API version {server_api}")
request = protocol.Request(id=next(request_ids))
query = request.execute_unparameterized_statement_request
query.language_name = "sql"
query.statement = "SELECT name FROM (VALUES ('Ada'), ('Lin')) AS example(name) ORDER BY name"
query.fetch_size = 1
statement = exchange(sock, request).statement_response
frame = statement.result.frame
while True:
for row in frame.relational_frame.rows:
print(row.values[0].string.string)
if frame.is_last:
break
request = protocol.Request(id=next(request_ids))
request.fetch_request.statement_id = statement.statement_id
request.fetch_request.fetch_size = 1
frame = exchange(sock, request).frame
request = protocol.Request(id=next(request_ids))
request.close_statement_request.statement_id = statement.statement_id
exchange(sock, request)
request = protocol.Request(id=next(request_ids))
request.disconnect_request.SetInParent()
exchange(sock, request)
The helper consumes intermediate responses and returns the final response. For direct execution, the intermediate response carries the statement ID before results arrive. To inspect a document namespace, send an EntitiesRequest on the same authenticated session and read the document variants of its EntitiesResponse. The protocol reference describes the available fields.
