Python sqlite3.Connection.blobopen Method

Last modified April 15, 2025

This comprehensive guide explores Python's sqlite3.Connection.blobopen method for efficient BLOB handling in SQLite databases. We'll cover basic usage, parameters, and practical examples.

Basic Definitions

The blobopen method provides direct access to BLOB data in SQLite without loading entire contents into memory. It returns a Blob object that supports file-like operations.

Key characteristics: it enables streaming of large BLOBs, supports random access, and works within transactions. The method requires a table name, column name, rowid, and optional read-only flag.

Basic BLOB Access

This example shows how to open and read a BLOB from a database table.

basic_blob.py

import sqlite3

with sqlite3.connect('images.db') as conn:
    # Create table and insert sample BLOB
    conn.execute('''CREATE TABLE IF NOT EXISTS images
                    (id INTEGER PRIMARY KEY, name TEXT, data BLOB)''')
    
    # Insert sample image (in real code, use actual binary data)
    conn.execute("INSERT INTO images (name, data) VALUES (?, ?)",
                 ('sample.png', b'PNG\x89\x50\x4E\x47\x0D\x0A\x1A\x0A'))
    
    # Get the last inserted rowid
    rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    
    # Open the BLOB
    blob = conn.blobopen('images', 'data', rowid)
    print(blob.read())  # Read entire BLOB
    blob.close()

This example creates a table with a BLOB column, inserts sample data, and reads it back using blobopen. The BLOB is accessed like a file object.

Note that we must know the rowid of the record containing the BLOB we want to access. The BLOB should be closed when done.

Reading BLOB in Chunks

For large BLOBs, reading in chunks is more memory-efficient than loading all at once.

chunked_read.py

import sqlite3

with sqlite3.connect('large_data.db') as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS documents
                    (id INTEGER PRIMARY KEY, name TEXT, content BLOB)''')
    
    # Insert large document (simulated)
    large_data = b'X' * (1024 * 1024)  # 1MB of data
    conn.execute("INSERT INTO documents (name, content) VALUES (?, ?)",
                 ('bigfile.bin', large_data))
    rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    
    # Read in 64KB chunks
    with conn.blobopen('documents', 'content', rowid) as blob:
        while chunk := blob.read(65536):  # 64KB chunks
            print(f"Read {len(chunk)} bytes")

This example demonstrates reading a large BLOB in manageable chunks. The with statement ensures the BLOB is properly closed after reading.

The chunk size (65536 bytes here) can be adjusted based on memory constraints and performance requirements.

Writing to a BLOB

The blobopen method can also be used to write BLOB data efficiently.

blob_write.py

import sqlite3

with sqlite3.connect('files.db') as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS files
                    (id INTEGER PRIMARY KEY, name TEXT, data BLOB)''')
    
    # Create empty record to get rowid
    conn.execute("INSERT INTO files (name) VALUES ('output.bin')")
    rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    
    # Write data to BLOB
    with conn.blobopen('files', 'data', rowid, readonly=False) as blob:
        blob.write(b'HEADER')
        blob.write(b'DATA' * 1000)
        blob.write(b'FOOTER')
    
    # Verify
    data = conn.execute("SELECT data FROM files WHERE id = ?", (rowid,)).fetchone()[0]
    print(f"Wrote {len(data)} bytes")

This example shows how to write data to a BLOB column. We first create an empty record to get a rowid, then open the BLOB in write mode.

The readonly=False parameter is required for writing. Changes are committed when the transaction commits.

Random Access in BLOBs

BLOB objects support random access using seek and tell methods, similar to file objects.

random_access.py

import sqlite3

with sqlite3.connect('random.db') as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS binary_data
                    (id INTEGER PRIMARY KEY, tag TEXT, payload BLOB)''')
    
    # Insert sample data
    data = bytes(range(256))  # 0-255 bytes
    conn.execute("INSERT INTO binary_data (tag, payload) VALUES (?, ?)",
                 ('sample', data))
    rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    
    # Access random positions
    with conn.blobopen('binary_data', 'payload', rowid) as blob:
        blob.seek(128)  # Go to middle
        print(f"Byte at 128: {blob.read(1)}")
        print(f"Current position: {blob.tell()}")
        blob.seek(-10, 2)  # 10 bytes from end
        print(f"Last 10 bytes: {blob.read()}")

This example demonstrates random access within a BLOB. The seek method moves the current position, and tell reports the current position.

The second parameter to seek specifies the reference point (0=start, 1=current, 2=end), similar to file operations.

Read-only BLOB Access

BLOBs can be opened in read-only mode for safety when only reading is needed.

readonly_blob.py

import sqlite3

with sqlite3.connect('products.db') as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS products
                    (id INTEGER PRIMARY KEY, name TEXT, image BLOB)''')
    
    # Insert sample product
    conn.execute("INSERT INTO products (name, image) VALUES (?, ?)",
                 ('Widget', b'IMAGE_DATA'))
    rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    
    # Open read-only
    with conn.blobopen('products', 'image', rowid, readonly=True) as blob:
        print(f"BLOB size: {blob.length()} bytes")
        try:
            blob.write(b'NEW')  # Will raise error
        except sqlite3.OperationalError as e:
            print(f"Expected error: {e}")

This example shows read-only BLOB access. Attempting to write raises an OperationalError. The length method returns the BLOB size.

Read-only mode is safer when you only need to read data and want to prevent accidental modifications.

BLOB with Explicit Transaction

BLOB operations participate in transactions, and changes are only permanent after commit.

blob_transaction.py

import sqlite3

with sqlite3.connect('transactions.db') as conn:
    conn.execute('''CREATE TABLE IF NOT EXISTS versions
                    (id INTEGER PRIMARY KEY, version INTEGER, data BLOB)''')
    
    # Start transaction
    conn.execute("BEGIN")
    try:
        # Insert and get rowid
        conn.execute("INSERT INTO versions (version) VALUES (1)")
        rowid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        
        # Write to BLOB
        with conn.blobopen('versions', 'data', rowid, readonly=False) as blob:
            blob.write(b'VERSION_1_DATA')
        
        # Verify before commit
        data = conn.execute("SELECT data FROM versions WHERE id = ?", (rowid,)).fetchone()
        print(f"Before commit: {data[0] is None}")  # True - not committed yet
        
        conn.commit()  # Make changes permanent
        
        # Verify after commit
        data = conn.execute("SELECT data FROM versions WHERE id = ?", (rowid,)).fetchone()
        print(f"After commit: {len(data[0])}")  # Length of written data
    except:
        conn.rollback()
        raise

This example demonstrates the transactional nature of BLOB operations. Changes are only visible to other connections after commit.

The explicit transaction ensures atomicity - either all changes succeed or none do. The BLOB write is part of this transaction.

BLOB with Row Factory

Combining BLOB access with row factories enables convenient data handling.

blob_row_factory.py

import sqlite3

def dict_factory(cursor, row):
    return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}

with sqlite3.connect('media.db') as conn:
    conn.row_factory = dict_factory
    conn.execute('''CREATE TABLE IF NOT EXISTS media
                    (id INTEGER PRIMARY KEY, title TEXT, content BLOB)''')
    
    # Insert sample media
    conn.execute("INSERT INTO media (title, content) VALUES (?, ?)",
                 ('Song', b'AUDIO_DATA'))
    
    # Query with row factory
    row = conn.execute("SELECT id, title FROM media").fetchone()
    print(f"Media title: {row['title']}")
    
    # Access BLOB using rowid from the same row
    with conn.blobopen('media', 'content', row['id']) as blob:
        print(f"First 10 bytes: {blob.read(10)}")

This example combines BLOB access with a row factory that returns dictionaries. We first query metadata, then use the rowid to access the BLOB content.

This pattern is useful when you need to examine metadata before deciding whether to load the potentially large BLOB content.

Best Practices

Always close BLOBs: Use context managers or explicit close
Use chunked reads: For large BLOBs to save memory
Consider read-only: When only reading is needed
Handle transactions: Remember BLOB writes need commit
Check row existence: Verify rowid before blobopen

Source References

Author

My name is Jan Bodnar, and I am a passionate programmer with extensive programming experience. I have been writing programming articles since 2007. To date, I have authored over 1,400 articles and 8 e-books. I possess more than ten years of experience in teaching programming.

List all Python tutorials.

Ebooks