summaryrefslogtreecommitdiff
path: root/python/database.py
diff options
context:
space:
mode:
authorKyleMiles <krm504@nyu.edu>2022-01-27 22:43:28 -0500
committerKyleMiles <krm504@nyu.edu>2022-01-28 00:24:06 -0500
commit6812c973c9fa9b4ad642ab81856c05f87bd6fcc4 (patch)
treedace4156d03148bcaf02df138ab4e0d93e61bc6f /python/database.py
parent519c9db22367f2659d1a54599fab47e6313be06e (diff)
Format All Files
Diffstat (limited to 'python/database.py')
-rw-r--r--python/database.py512
1 files changed, 256 insertions, 256 deletions
diff --git a/python/database.py b/python/database.py
index a563d99c..b9b81058 100644
--- a/python/database.py
+++ b/python/database.py
@@ -28,322 +28,322 @@ from . import filemetadata
class KeyValueStore:
- """
+ """
``class KeyValueStore`` maintains access to the raw data stored in Snapshots and various
other Database-related structures.
"""
- def __init__(self, buffer: Optional[databuffer.DataBuffer] = None, handle=None):
- if handle is not None:
- self.handle = core.handle_of_type(handle, core.BNKeyValueStore)
- else:
- if buffer is None:
- _handle = core.BNCreateKeyValueStore()
- else:
- _handle = core.BNCreateKeyValueStoreFromDataBuffer(buffer.handle)
- assert _handle is not None
- self.handle = _handle
+ def __init__(self, buffer: Optional[databuffer.DataBuffer] = None, handle=None):
+ if handle is not None:
+ self.handle = core.handle_of_type(handle, core.BNKeyValueStore)
+ else:
+ if buffer is None:
+ _handle = core.BNCreateKeyValueStore()
+ else:
+ _handle = core.BNCreateKeyValueStoreFromDataBuffer(buffer.handle)
+ assert _handle is not None
+ self.handle = _handle
- def __del__(self):
- core.BNFreeKeyValueStore(self.handle)
+ def __del__(self):
+ core.BNFreeKeyValueStore(self.handle)
- def __getitem__(self, item: str) -> databuffer.DataBuffer:
- return self.get_value(item)
+ def __getitem__(self, item: str) -> databuffer.DataBuffer:
+ return self.get_value(item)
- def __setitem__(self, key: str, value: databuffer.DataBuffer):
- return self.set_value(key, value)
+ def __setitem__(self, key: str, value: databuffer.DataBuffer):
+ return self.set_value(key, value)
- @property
- def keys(self):
- """Get a list of all keys stored in the kvs (read-only)"""
- count = ctypes.c_ulonglong(0)
- value = core.BNGetKeyValueStoreKeys(self.handle, count)
- assert value is not None
+ @property
+ def keys(self):
+ """Get a list of all keys stored in the kvs (read-only)"""
+ count = ctypes.c_ulonglong(0)
+ value = core.BNGetKeyValueStoreKeys(self.handle, count)
+ assert value is not None
- result = []
- try:
- for i in range(0, count.value):
- result.append(value[i])
- return result
- finally:
- core.BNFreeStringList(value, count)
+ result = []
+ try:
+ for i in range(0, count.value):
+ result.append(value[i])
+ return result
+ finally:
+ core.BNFreeStringList(value, count)
- def get_value(self, key: str) -> databuffer.DataBuffer:
- """Get the value for a single key"""
- handle = core.BNGetKeyValueStoreBuffer(self.handle, key)
- assert handle is not None
- return databuffer.DataBuffer(handle=handle)
+ def get_value(self, key: str) -> databuffer.DataBuffer:
+ """Get the value for a single key"""
+ handle = core.BNGetKeyValueStoreBuffer(self.handle, key)
+ assert handle is not None
+ return databuffer.DataBuffer(handle=handle)
- def set_value(self, key: str, value: databuffer.DataBuffer):
- """Set the value for a single key"""
- core.BNSetKeyValueStoreBuffer(self.handle, key, value.handle)
+ def set_value(self, key: str, value: databuffer.DataBuffer):
+ """Set the value for a single key"""
+ core.BNSetKeyValueStoreBuffer(self.handle, key, value.handle)
- @property
- def serialized_data(self) -> databuffer.DataBuffer:
- """Get the stored representation of the kvs (read-only)"""
- handle = core.BNGetKeyValueStoreSerializedData(self.handle)
- assert handle is not None
- return databuffer.DataBuffer(handle=handle)
+ @property
+ def serialized_data(self) -> databuffer.DataBuffer:
+ """Get the stored representation of the kvs (read-only)"""
+ handle = core.BNGetKeyValueStoreSerializedData(self.handle)
+ assert handle is not None
+ return databuffer.DataBuffer(handle=handle)
- def begin_namespace(self, name: str):
- """Begin storing new keys into a namespace"""
- core.BNBeginKeyValueStoreNamespace(self.handle, name)
+ def begin_namespace(self, name: str):
+ """Begin storing new keys into a namespace"""
+ core.BNBeginKeyValueStoreNamespace(self.handle, name)
- def end_namespace(self):
- """End storing new keys into a namespace"""
- core.BNEndKeyValueStoreNamespace(self.handle)
+ def end_namespace(self):
+ """End storing new keys into a namespace"""
+ core.BNEndKeyValueStoreNamespace(self.handle)
- @property
- def empty(self) -> bool:
- """If the kvs is empty (read-only)"""
- return core.BNIsKeyValueStoreEmpty(self.handle)
+ @property
+ def empty(self) -> bool:
+ """If the kvs is empty (read-only)"""
+ return core.BNIsKeyValueStoreEmpty(self.handle)
- @property
- def value_size(self) -> bool:
- """Number of values in the kvs (read-only)"""
- return core.BNGetKeyValueStoreValueSize(self.handle)
+ @property
+ def value_size(self) -> bool:
+ """Number of values in the kvs (read-only)"""
+ return core.BNGetKeyValueStoreValueSize(self.handle)
- @property
- def data_size(self) -> bool:
- """Length of serialized data (read-only)"""
- return core.BNGetKeyValueStoreDataSize(self.handle)
+ @property
+ def data_size(self) -> bool:
+ """Length of serialized data (read-only)"""
+ return core.BNGetKeyValueStoreDataSize(self.handle)
- @property
- def value_storage_size(self) -> bool:
- """Size of all data in storage (read-only)"""
- return core.BNGetKeyValueStoreValueStorageSize(self.handle)
+ @property
+ def value_storage_size(self) -> bool:
+ """Size of all data in storage (read-only)"""
+ return core.BNGetKeyValueStoreValueStorageSize(self.handle)
- @property
- def namespace_size(self) -> bool:
- """Number of namespaces pushed with begin_namespace (read-only)"""
- return core.BNGetKeyValueStoreNamespaceSize(self.handle)
+ @property
+ def namespace_size(self) -> bool:
+ """Number of namespaces pushed with begin_namespace (read-only)"""
+ return core.BNGetKeyValueStoreNamespaceSize(self.handle)
class Snapshot:
- """
+ """
``class Snapshot`` is a model of an individual database snapshot, created on save.
"""
- def __init__(self, handle):
- self.handle = core.handle_of_type(handle, core.BNSnapshot)
+ def __init__(self, handle):
+ self.handle = core.handle_of_type(handle, core.BNSnapshot)
- def __del__(self):
- core.BNFreeSnapshot(self.handle)
+ def __del__(self):
+ core.BNFreeSnapshot(self.handle)
- @property
- def database(self) -> 'Database':
- """Get the owning database (read-only)"""
- return Database(handle=core.BNGetSnapshotDatabase(self.handle))
+ @property
+ def database(self) -> 'Database':
+ """Get the owning database (read-only)"""
+ return Database(handle=core.BNGetSnapshotDatabase(self.handle))
- @property
- def id(self) -> int:
- """Get the numerical id (read-only)"""
- return core.BNGetSnapshotId(self.handle)
+ @property
+ def id(self) -> int:
+ """Get the numerical id (read-only)"""
+ return core.BNGetSnapshotId(self.handle)
- @property
- def name(self) -> str:
- """Get the displayed snapshot name (read-only)"""
- return core.BNGetSnapshotName(self.handle)
+ @property
+ def name(self) -> str:
+ """Get the displayed snapshot name (read-only)"""
+ return core.BNGetSnapshotName(self.handle)
- @property
- def is_auto_save(self) -> bool:
- """If the snapshot was the result of an auto-save (read-only)"""
- return core.BNIsSnapshotAutoSave(self.handle)
+ @property
+ def is_auto_save(self) -> bool:
+ """If the snapshot was the result of an auto-save (read-only)"""
+ return core.BNIsSnapshotAutoSave(self.handle)
- @property
- def has_contents(self) -> bool:
- """If the snapshot has contents, and has not been trimmed (read-only)"""
- return core.BNSnapshotHasContents(self.handle)
+ @property
+ def has_contents(self) -> bool:
+ """If the snapshot has contents, and has not been trimmed (read-only)"""
+ return core.BNSnapshotHasContents(self.handle)
- @property
- def has_undo(self) -> bool:
- """If the snapshot has undo data (read-only)"""
- return core.BNSnapshotHasUndo(self.handle)
+ @property
+ def has_undo(self) -> bool:
+ """If the snapshot has undo data (read-only)"""
+ return core.BNSnapshotHasUndo(self.handle)
- @property
- def first_parent(self) -> Optional['Snapshot']:
- """Get the first parent of the snapshot, or None if it has no parents (read-only)"""
- handle = core.BNGetSnapshotFirstParent(self.handle)
- if handle is None:
- return None
- return Snapshot(handle=handle)
+ @property
+ def first_parent(self) -> Optional['Snapshot']:
+ """Get the first parent of the snapshot, or None if it has no parents (read-only)"""
+ handle = core.BNGetSnapshotFirstParent(self.handle)
+ if handle is None:
+ return None
+ return Snapshot(handle=handle)
- @property
- def parents(self) -> List['Snapshot']:
- """Get a list of all parent snapshots of the snapshot (read-only)"""
- count = ctypes.c_ulonglong(0)
- parents = core.BNGetSnapshotParents(self.handle, count)
+ @property
+ def parents(self) -> List['Snapshot']:
+ """Get a list of all parent snapshots of the snapshot (read-only)"""
+ count = ctypes.c_ulonglong(0)
+ parents = core.BNGetSnapshotParents(self.handle, count)
- result = []
- try:
- for i in range(0, count.value):
- handle = core.BNNewSnapshotReference(parents[i])
- result.append(Snapshot(handle=handle))
- return result
- finally:
- core.BNFreeSnapshotList(parents, count)
+ result = []
+ try:
+ for i in range(0, count.value):
+ handle = core.BNNewSnapshotReference(parents[i])
+ result.append(Snapshot(handle=handle))
+ return result
+ finally:
+ core.BNFreeSnapshotList(parents, count)
- @property
- def children(self) -> List['Snapshot']:
- """Get a list of all child snapshots of the snapshot (read-only)"""
- count = ctypes.c_ulonglong(0)
- children = core.BNGetSnapshotChildren(self.handle, count)
+ @property
+ def children(self) -> List['Snapshot']:
+ """Get a list of all child snapshots of the snapshot (read-only)"""
+ count = ctypes.c_ulonglong(0)
+ children = core.BNGetSnapshotChildren(self.handle, count)
- result = []
- try:
- for i in range(0, count.value):
- handle = core.BNNewSnapshotReference(children[i])
- result.append(Snapshot(handle=handle))
- return result
- finally:
- core.BNFreeSnapshotList(children, count)
+ result = []
+ try:
+ for i in range(0, count.value):
+ handle = core.BNNewSnapshotReference(children[i])
+ result.append(Snapshot(handle=handle))
+ return result
+ finally:
+ core.BNFreeSnapshotList(children, count)
- @property
- def file_contents(self) -> databuffer.DataBuffer:
- """Get a buffer of the raw data at the time of the snapshot (read-only)"""
- assert self.has_contents
- handle = core.BNGetSnapshotFileContents(self.handle)
- assert handle is not None
- return databuffer.DataBuffer(handle=handle)
+ @property
+ def file_contents(self) -> databuffer.DataBuffer:
+ """Get a buffer of the raw data at the time of the snapshot (read-only)"""
+ assert self.has_contents
+ handle = core.BNGetSnapshotFileContents(self.handle)
+ assert handle is not None
+ return databuffer.DataBuffer(handle=handle)
- @property
- def file_contents_hash(self) -> databuffer.DataBuffer:
- """Get a hash of the data at the time of the snapshot (read-only)"""
- assert self.has_contents
- handle = core.BNGetSnapshotFileContentsHash(self.handle)
- assert handle is not None
- return databuffer.DataBuffer(handle=handle)
+ @property
+ def file_contents_hash(self) -> databuffer.DataBuffer:
+ """Get a hash of the data at the time of the snapshot (read-only)"""
+ assert self.has_contents
+ handle = core.BNGetSnapshotFileContentsHash(self.handle)
+ assert handle is not None
+ return databuffer.DataBuffer(handle=handle)
- @property
- def undo_entries(self):
- """Get a list of undo entries at the time of the snapshot (read-only)"""
- assert self.has_undo
- raise NotImplementedError("GetUndoEntries is not implemented in python")
+ @property
+ def undo_entries(self):
+ """Get a list of undo entries at the time of the snapshot (read-only)"""
+ assert self.has_undo
+ raise NotImplementedError("GetUndoEntries is not implemented in python")
- @property
- def data(self) -> KeyValueStore:
- """Get the backing kvs data with snapshot fields (read-only)"""
- assert self.has_contents
- handle = core.BNReadSnapshotData(self.handle)
- assert handle is not None
- return KeyValueStore(handle=handle)
+ @property
+ def data(self) -> KeyValueStore:
+ """Get the backing kvs data with snapshot fields (read-only)"""
+ assert self.has_contents
+ handle = core.BNReadSnapshotData(self.handle)
+ assert handle is not None
+ return KeyValueStore(handle=handle)
- def has_ancestor(self, other: 'Snapshot') -> bool:
- """Determine if this snapshot has another as an ancestor"""
- return core.BNSnapshotHasAncestor(self.handle, other.handle)
+ def has_ancestor(self, other: 'Snapshot') -> bool:
+ """Determine if this snapshot has another as an ancestor"""
+ return core.BNSnapshotHasAncestor(self.handle, other.handle)
class Database:
- """
+ """
``class Database`` provides lower level access to raw snapshot data used to construct analysis data
"""
- def __init__(self, handle):
- self.handle = core.handle_of_type(handle, core.BNDatabase)
+ def __init__(self, handle):
+ self.handle = core.handle_of_type(handle, core.BNDatabase)
- def __del__(self):
- core.BNFreeDatabase(self.handle)
+ def __del__(self):
+ core.BNFreeDatabase(self.handle)
- def __getitem__(self, item: int) -> Optional[Snapshot]:
- return self.get_snapshot(item)
+ def __getitem__(self, item: int) -> Optional[Snapshot]:
+ return self.get_snapshot(item)
- def get_snapshot(self, id: int) -> Optional[Snapshot]:
- """Get a snapshot by its id, or None if no snapshot with that id exists"""
- snap = core.BNGetDatabaseSnapshot(self.handle, id)
- if snap is None:
- return None
- return Snapshot(handle=snap)
+ def get_snapshot(self, id: int) -> Optional[Snapshot]:
+ """Get a snapshot by its id, or None if no snapshot with that id exists"""
+ snap = core.BNGetDatabaseSnapshot(self.handle, id)
+ if snap is None:
+ return None
+ return Snapshot(handle=snap)
- @property
- def snapshots(self) -> List[Snapshot]:
- """Get a list of all snapshots in the database (read-only)"""
- count = ctypes.c_ulonglong(0)
- snapshots = core.BNGetDatabaseSnapshots(self.handle, count)
- assert snapshots is not None
+ @property
+ def snapshots(self) -> List[Snapshot]:
+ """Get a list of all snapshots in the database (read-only)"""
+ count = ctypes.c_ulonglong(0)
+ snapshots = core.BNGetDatabaseSnapshots(self.handle, count)
+ assert snapshots is not None
- result = []
- try:
- for i in range(0, count.value):
- handle = core.BNNewSnapshotReference(snapshots[i])
- result.append(Snapshot(handle=handle))
- return result
- finally:
- core.BNFreeSnapshotList(snapshots, count)
+ result = []
+ try:
+ for i in range(0, count.value):
+ handle = core.BNNewSnapshotReference(snapshots[i])
+ result.append(Snapshot(handle=handle))
+ return result
+ finally:
+ core.BNFreeSnapshotList(snapshots, count)
- @property
- def current_snapshot(self) -> Optional[Snapshot]:
- """Get the current snapshot"""
- snap = core.BNGetDatabaseCurrentSnapshot(self.handle)
- if snap is None:
- return None
- return Snapshot(handle=snap)
+ @property
+ def current_snapshot(self) -> Optional[Snapshot]:
+ """Get the current snapshot"""
+ snap = core.BNGetDatabaseCurrentSnapshot(self.handle)
+ if snap is None:
+ return None
+ return Snapshot(handle=snap)
- @current_snapshot.setter
- def current_snapshot(self, value: Snapshot):
- core.BNSetDatabaseCurrentSnapshot(self.handle, value.id)
+ @current_snapshot.setter
+ def current_snapshot(self, value: Snapshot):
+ core.BNSetDatabaseCurrentSnapshot(self.handle, value.id)
- def remove_snapsot(self, id: int):
- """Remove a snapshot in the database by id"""
- core.BNRemoveDatabaseSnapshot(self.handle, id)
+ def remove_snapsot(self, id: int):
+ """Remove a snapshot in the database by id"""
+ core.BNRemoveDatabaseSnapshot(self.handle, id)
- @property
- def global_keys(self) -> List[str]:
- """Get a list of keys for all globals in the database (read-only)"""
- count = ctypes.c_ulonglong(0)
- value = core.BNGetDatabaseGlobalKeys(self.handle, count)
- assert value is not None
+ @property
+ def global_keys(self) -> List[str]:
+ """Get a list of keys for all globals in the database (read-only)"""
+ count = ctypes.c_ulonglong(0)
+ value = core.BNGetDatabaseGlobalKeys(self.handle, count)
+ assert value is not None
- result = []
- try:
- for i in range(0, count.value):
- result.append(value[i])
- return result
- finally:
- core.BNFreeStringList(value, count)
+ result = []
+ try:
+ for i in range(0, count.value):
+ result.append(value[i])
+ return result
+ finally:
+ core.BNFreeStringList(value, count)
- @property
- def globals(self) -> Dict[str, str]:
- """Get a dictionary of all globals (read-only)"""
- count = ctypes.c_ulonglong(0)
- value = core.BNGetDatabaseGlobalKeys(self.handle, count)
- assert value is not None
+ @property
+ def globals(self) -> Dict[str, str]:
+ """Get a dictionary of all globals (read-only)"""
+ count = ctypes.c_ulonglong(0)
+ value = core.BNGetDatabaseGlobalKeys(self.handle, count)
+ assert value is not None
- result = {}
- try:
- for i in range(0, count.value):
- key = value[i]
- result[key] = self.read_global(key)
- return result
- finally:
- core.BNFreeStringList(value, count)
+ result = {}
+ try:
+ for i in range(0, count.value):
+ key = value[i]
+ result[key] = self.read_global(key)
+ return result
+ finally:
+ core.BNFreeStringList(value, count)
- def read_global(self, key: str) -> str:
- """Get a specific global by key"""
- value = core.BNReadDatabaseGlobal(self.handle, key)
- assert value is not None
- return value
+ def read_global(self, key: str) -> str:
+ """Get a specific global by key"""
+ value = core.BNReadDatabaseGlobal(self.handle, key)
+ assert value is not None
+ return value
- def write_global(self, key: str, value: str):
- """Write a global into the database"""
- core.BNWriteDatabaseGlobal(self.handle, key, value)
+ def write_global(self, key: str, value: str):
+ """Write a global into the database"""
+ core.BNWriteDatabaseGlobal(self.handle, key, value)
- def read_global_data(self, key: str) -> databuffer.DataBuffer:
- """Get a specific global by key, as a binary buffer"""
- handle = core.BNReadDatabaseGlobalData(self.handle, key)
- assert handle is not None
- return databuffer.DataBuffer(handle=handle)
+ def read_global_data(self, key: str) -> databuffer.DataBuffer:
+ """Get a specific global by key, as a binary buffer"""
+ handle = core.BNReadDatabaseGlobalData(self.handle, key)
+ assert handle is not None
+ return databuffer.DataBuffer(handle=handle)
- def write_global_data(self, key: str, value: databuffer.DataBuffer):
- """Write a binary buffer into a global in the database"""
- core.BNWriteDatabaseGlobalData(self.handle, key, value.handle)
+ def write_global_data(self, key: str, value: databuffer.DataBuffer):
+ """Write a binary buffer into a global in the database"""
+ core.BNWriteDatabaseGlobalData(self.handle, key, value.handle)
- @property
- def file(self) -> 'filemetadata.FileMetadata':
- """Get the owning FileMetadata (read-only)"""
- handle = core.BNGetDatabaseFile(self.handle)
- assert handle is not None
- return filemetadata.FileMetadata(handle=handle)
+ @property
+ def file(self) -> 'filemetadata.FileMetadata':
+ """Get the owning FileMetadata (read-only)"""
+ handle = core.BNGetDatabaseFile(self.handle)
+ assert handle is not None
+ return filemetadata.FileMetadata(handle=handle)
- @property
- def analysis_cache(self) -> KeyValueStore:
- """Get the backing analysis cache kvs (read-only)"""
- handle = core.BNReadDatabaseAnalysisCache(self.handle)
- assert handle is not None
- return KeyValueStore(handle=handle)
+ @property
+ def analysis_cache(self) -> KeyValueStore:
+ """Get the backing analysis cache kvs (read-only)"""
+ handle = core.BNReadDatabaseAnalysisCache(self.handle)
+ assert handle is not None
+ return KeyValueStore(handle=handle)