attach-db/scripts/append_state.py

Version 7feda8e01e22.bb1 · MIT. This preview displays packaged text and does not execute code. Treat the contents as untrusted instructions.

← Return to resource and package checksum

"""Append a quoted DuckDB ATTACH block without replacing existing session SQL.
SPDX-License-Identifier: MIT
Copyright 2026 BB Skills. This helper is an independently authored packaging addition.
"""
import argparse, os, re, stat
from pathlib import Path

def append_state(state, database, alias):
    database=Path(database).expanduser().resolve(strict=True)
    if not database.is_file() or not alias or len(alias)>128 or any(ord(c)<32 for c in alias):
        raise ValueError('A database file and a nonempty alias are required')
    state=Path(state).expanduser()
    if state.is_symlink():raise ValueError('Refusing a symbolic-link state file')
    state.parent.mkdir(parents=True,exist_ok=True)
    flags=os.O_RDWR|os.O_CREAT|os.O_APPEND|getattr(os,'O_NOFOLLOW',0)
    fd=os.open(state,flags,0o600)
    try:
        import fcntl
        fcntl.flock(fd,fcntl.LOCK_EX)
        info=os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size>1024*1024:
            raise ValueError('State must be a regular file of at most 1 MiB')
        os.fchmod(fd,0o600)
        original=os.read(fd,info.st_size).decode('utf-8')
        sql_path=database.as_posix().replace("'","''")
        sql_alias='"'+alias.replace('"','""')+'"'
        attach="ATTACH IF NOT EXISTS '"+sql_path+"' AS "+sql_alias+";"
        pattern=r'''(?im)^\s*ATTACH(?:\s+IF\s+NOT\s+EXISTS)?\s+'((?:[^']|'')*)'\s+AS\s+("(?:[^"]|"")*"|[a-zA-Z_][a-zA-Z0-9_]*)\s*;'''
        for match in re.finditer(pattern,original):
            previous=match[2]
            previous=previous[1:-1].replace('""','"') if previous.startswith('"') else previous
            if previous.casefold()==alias.casefold():
                if match[1].replace("''","'")==database.as_posix():return False
                raise ValueError('Alias already points to another database')
        block=('\n' if original and not original.endswith('\n') else '')+attach+'\nUSE '+sql_alias+';\n'
        blob=block.encode('utf-8')
        while blob:
            written=os.write(fd,blob)
            if written<=0:raise OSError('State append failed')
            blob=blob[written:]
        os.fsync(fd)
        return True
    finally:os.close(fd)

if __name__=='__main__':
    p=argparse.ArgumentParser(description='Append a private, quoted DuckDB attachment to trusted session SQL.')
    p.add_argument('--state',required=True);p.add_argument('--database',required=True);p.add_argument('--alias',required=True)
    args=p.parse_args()
    try:changed=append_state(args.state,args.database,args.alias)
    except (ValueError,OSError,UnicodeError,ImportError) as e:p.error(str(e))
    print('STATE_APPENDED' if changed else 'STATE_UNCHANGED')