Ich stecke mit dem Modul APSW eine VFS für SQLite in Python implementierenPython

Python-Programme
Anonymous
 Ich stecke mit dem Modul APSW eine VFS für SQLite in Python implementieren

Post by Anonymous »

Ich habe eine Weile an der Implementierung eines benutzerdefinierten VFS für SQLite3 in Python mit dem APSW -Modul mit dem Ziel der Verwaltung einer In-Memory-Datenbank gearbeitet. Ich weiß, dass SQLITE integrierte Unterstützung für In-Memory-Datenbanken hat, aber soweit ich weiß, ermöglicht es nicht, eine vorhandene Datenbank in den Speicher zu importieren-es sei denn, ich habe etwas verpasst.

Code: Select all

with open('database.db', 'rb') as f:
db_bytes = f.read()

< /code>
Ich führe dann alle notwendigen VFS -Vorgänge aus (Xread, XWRITE usw.). Es funktioniert ganz gut: Ich kann eine SQL -Verbindung mit einem Bytesio -Puffer ohne Probleme lesen und schreiben. Wenn ich den Bytesio -Puffer zurück auf die Festplatte schreibe und einen Hexdump ausführt, sehe ich, dass die Tabelle dupliziert erscheint. < /P>
Hier ist eine vereinfachte Version meiner Implementierung:class VolatileFS(apsw.VFS):
def __init__(self, db_bytes, vfsname='volatilefs', basevfs=''):
self.vfs_name = vfsname
self.base_vfs = basevfs
self.db = BytesIO(db_bytes)
super().__init__(self.vfs_name, self.base_vfs)

def xOpen(self, name: str, flags: Iterable[int]):
self.file = VolatileFSFile(self.db, self.base_vfs, name, flags)
return self.file

class VolatileFSFile(apsw.VFSFile):
def __init__(self, db, inheritfromvfsname, filename, flags):
self.buffer = db
super().__init__(inheritfromvfsname, filename, flags)

def xClose(self):
super().xClose()

def xRead(self, amount: int, offset: int):
self.buffer.seek(offset)
data = self.buffer.read(amount)
if len(data) < amount:
data += b'\x00' * (amount - len(data))
return data
#return super().xRead(amount, offset)

def xWrite(self, data: bytes, offset: int):
if len(self.buffer.getvalue()) < offset + len(data):
size = offset + len(data) - len(self.buffer.getvalue())
self.buffer.seek(0, 2)
self.buffer.write(b'\x00' * size)
self.buffer.seek(offset)
self.buffer.write(data)

def xTruncate(self, size: int):
current = self.buffer.getvalue()
self.buffer = current[:size]

def xFileSize(self) -> int:
return len(self.buffer.getvalue())
Wenn Sie interessiert sind, habe ich drei Puffer -Dumps im Repo unter https://github.com/blantheon/volatilefs:
hinzugefügt

Code: Select all

buffer1: initial state (only contains a dropped empty table)

buffer2: after creating the actual table

buffer3: after inserting values (this is where duplication appears)
Zum Vergleich enthält die test_sqlite -Datei genau die gleichen SQL -Operationen, die mit normalem SQLite ausgeführt wurden.>

Quick Reply

Change Text Case: 
   
  • Similar Topics
    Replies
    Views
    Last post