Difference between revisions of "Python Sample SQLite"
Jump to navigation
Jump to search
(→APSW) |
|||
| Line 12: | Line 12: | ||
You can verify that it is working by running the following script: | You can verify that it is working by running the following script: | ||
<source lang="python"> | <source lang="python"> | ||
| − | |||
import sys | import sys | ||
| − | |||
import apsw | import apsw | ||
import apsw.ext | import apsw.ext | ||
| − | |||
| − | |||
| − | |||
def main() -> None: | def main() -> None: | ||
Revision as of 05:45, 1 March 2024
It is possible to access SQLite files in Truxton without first exporting them to your local system.
APSW
Since the Python DB-API only supports establishing a connection to a SQLite database by providing a file name, we have to create our own filesystem to read from the file in Truxton. SQLite has the concept of virtual filesystems but DB-API doesn't support it. However, another SQLite library called APSW does. Install it using pip:
pip install apsw
You can verify that it is working by running the following script:
import sys
import apsw
import apsw.ext
def main() -> None:
print(" Using APSW file", apsw.__file__)
print(" APSW version", apsw.apsw_version())
print("SQLite header version", apsw.SQLITE_VERSION_NUMBER)
print(" SQLite lib version", apsw.sqlite_lib_version())
print(" Using amalgamation", apsw.using_amalgamation)
return None
if __name__ == "__main__":
sys.exit(main())
SQLite in Truxton
Here's a script to query a SQLite database that is stored in Truxton.
import sys
import os
import apsw
from pathlib import Path
sys.path.append('C:/Program Files/Truxton/SDK')
import truxton
def main() -> None:
truxton_virtual_filesystem = TruxtonVFS()
truxton_virtual_filesystem.Truxton = truxton.create()
database = apsw.Connection( "65d9d912-7328-d2e7-d511-f909000000c6", flags = SQLITE_OPEN_READONLY, vfs = "truxton")
for row in database.execute("SELECT Z_UUID FROM Z_METADATA"):
print(row)
database.close()
return None
SQLITE_OPEN_READONLY = 0x00000001
class TruxtonVFS(apsw.VFS):
Truxton = None
def __init__(self, vfsname="truxton", basevfs=""):
self.vfs_name = vfsname
self.base_vfs = basevfs
super().__init__(self.vfs_name, self.base_vfs)
def xOpen(self, fname, flags):
# fname is of type apsw.URIFilename object
truxton_guid = os.path.basename(fname.filename())
opened_file = self.Truxton.getfileid(truxton_guid)
return TruxtonVFSFile(f = opened_file, name = truxton_guid, flags = flags)
def xAccess(self, pathname, flags):
return False
def xDelete(self, filename, syncdir):
pass
class TruxtonVFSFile(apsw.VFSFile):
def __init__(self, f: truxton.TruxtonFileIO, name, flags):
"""
Truxton VFS File object
Args:
* f: TruxtonFileIO object returned by Truxton.getfileid
* name: name of the file
* flags: SQLite open flags
"""
self.f = f
self.flags = flags
self.name = name
self.mode = "rb"
def xRead(self, amount, offset) -> bytes:
self.f.seek(offset)
data = self.f.read(amount)
return data
def xFileControl(self, *args):
return True
def xDeviceCharacteristics(self):
return 4096
def xCheckReservedLock(self):
return False
def xLock(self, level):
pass
def xUnlock(self, level):
pass
def xSectorSize(self):
return self.f.block_size
def xClose(self):
self.f.close()
pass
def xFileSize(self):
pos = self.f.tell()
self.f.seek(0, 2)
size = self.f.tell()
self.f.seek(pos)
return size
def xSync(self, flags):
pass
def xTruncate(self, newsize):
pass
def xWrite(self, data, offset):
pass
if __name__ == "__main__":
sys.exit(main())