[Python] I'm stuck in implementing a VFS for sqlite in python with the module apsw

Discussão em 'Python' iniciado por Stack, Maio 9, 2025.

  1. Stack

    Stack Membro Participativo

    I've been working for a while on implementing a custom VFS for SQLite3 in Python using the apsw module, with the goal of managing an in-memory database. I know that SQLite has built-in support for in-memory databases, but as far as I understand, it doesn’t allow importing an existing database into memory — unless I missed something.

    My VFS takes a db_bytes object (a bytes object obtained via):

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



    I then perform all the necessary VFS operations on it (xRead, xWrite, etc.). It works quite well: I can read and write through a SQL connection using a BytesIO buffer with no issues.

    However, I encounter a problem after inserting data into a table. When I write the BytesIO buffer back to disk and run a hexdump, I see that the table appears duplicated.

    Here’s a simplified version of my implementation:

    class VolatileFS(apsw.VFS):
    def __init__(self, db_bytes, vfsname='volatilefs', basevfs=''):
    self.vfs_name = vfsname
    self.base_vfs = basevfs
    self.db: bytes = 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())


    If you're interested, I’ve added three buffer dumps in the repo at https://github.com/Blantheon/VolatileFS:

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

    buffer2: after creating the actual table

    buffer3: after inserting values (this is where duplication appears)


    For comparison, the test_sqlite file contains the exact same SQL operations performed using regular SQLite.

    I've also included a script called stack_overflow_help.sh to reproduce the issue and illustrate what I'm trying to explain in this post.

    Any help or guidance would be appreciated!

    Continue reading...

Compartilhe esta Página