Skip to content

Instantly share code, notes, and snippets.

@Razikus
Last active July 6, 2022 17:33
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save Razikus/3b25067a83c32f498422a13508711f39 to your computer and use it in GitHub Desktop.
Save Razikus/3b25067a83c32f498422a13508711f39 to your computer and use it in GitHub Desktop.
immudb example for creating table with index
from immudb import ImmudbClient
import time
class ExampleClient:
def __init__(self, immudbUrl, login, password, database):
self.immudbUrl = immudbUrl
self.login = login
self.password = password
self.database = database
self.client = ImmudbClient(immudbUrl)
self.lastLogin = -1
def __enter__(self):
if(self.lastLogin + 60 * 10 < time.time()):
self.client.login(self.login, self.password, self.database)
self.lastLogin = time.time()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
pass
def insertNewRecord(self, value: str):
with self:
query = f"INSERT INTO example_table (exampleField) VALUES(@what)"
return self.client.sqlExec(query, {"what": value})
def getRecordById(self, id: int):
with self:
query = f"SELECT id, exampleField from example_table WHERE id = @id"
return self.client.sqlQuery(query, {"id": id})
def createTables(self):
with self:
query = f"""CREATE TABLE IF NOT EXISTS example_table (
id INTEGER AUTO_INCREMENT,
exampleField VARCHAR[60],
PRIMARY KEY (id)
);
CREATE INDEX IF NOT EXISTS on example_table (exampleField);"""
self.client.sqlExec(query, dict())
if __name__ == "__main__":
example = ExampleClient("localhost:3322", "immudb", "immudb", "defaultdb")
example.createTables()
returned = example.insertNewRecord("test")
print("Transaction number", returned.txs[0].header.id)
queried = example.getRecordById(1)
if(len(queried) > 0):
idOf, valueOf = queried[0]
print(f"Found record with id '{idOf}' and value '{valueOf}'")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment