90 lines
3.8 KiB
Python
90 lines
3.8 KiB
Python
"""Native transactional writes reusing the shared identity/provenance policy."""
|
|
import re
|
|
import zlib
|
|
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.postgres import text
|
|
from paper_atlas.postgres.client import connect
|
|
from paper_atlas.postgres.schema import TABLES, quote
|
|
|
|
|
|
class Result:
|
|
def __init__(self, cursor):
|
|
self.cursor = cursor
|
|
|
|
def fetchone(self):
|
|
row = self.cursor.fetchone()
|
|
return tuple(map(text.decode, row)) if row is not None else None
|
|
|
|
def fetchall(self):
|
|
return [tuple(map(text.decode, row)) for row in self.cursor.fetchall()]
|
|
|
|
def __iter__(self):
|
|
return (tuple(map(text.decode, row)) for row in self.cursor)
|
|
|
|
@property
|
|
def rowcount(self):
|
|
return self.cursor.rowcount
|
|
|
|
|
|
class WriteConnection:
|
|
postgres = True
|
|
|
|
def __init__(self, service, schema=None):
|
|
self.connection = connect(service)
|
|
try:
|
|
if schema is None:
|
|
schema = self.connection.execute('SELECT current_schema()').fetchone()[0]
|
|
if not isinstance(schema, str) or not re.fullmatch(r'[a-z][a-z0-9_]{0,35}', schema):
|
|
raise ValueError('Invalid PostgreSQL write schema')
|
|
annotation = self.connection.execute("SELECT obj_description(oid,'pg_namespace') FROM pg_namespace WHERE nspname=%s", (schema,)).fetchone()
|
|
if not annotation or annotation[0] != 'paper-atlas:text-codec=' + text.CODEC:
|
|
raise RuntimeError('Writes require a published dataset with the supported text codec')
|
|
tables = {r[0] for r in self.connection.execute("SELECT table_name FROM information_schema.tables WHERE table_schema=%s AND table_type='BASE TABLE'", (schema,))}
|
|
if not set(TABLES) <= tables:
|
|
raise RuntimeError('Published business tables are incomplete or inaccessible')
|
|
if not self.connection.execute('SELECT pg_try_advisory_lock(%s,%s)', (615174, zlib.crc32(schema.encode()) & 0x7fffffff)).fetchone()[0]:
|
|
raise RuntimeError('Another writer holds this PostgreSQL dataset; retry after it closes')
|
|
self.schema = schema
|
|
self.connection.execute('SET search_path TO ' + quote(schema) + ',pg_catalog')
|
|
if not self.connection.execute("SELECT 1 FROM schema_migrations WHERE name='submission_source_baseline_v1'").fetchone():
|
|
raise RuntimeError('Submission provenance baseline is missing')
|
|
self.connection.autocommit = False
|
|
except BaseException:
|
|
self.connection.close()
|
|
raise
|
|
|
|
def execute(self, sql, params=()):
|
|
# The shared domain uses static SQL and bound parameters. It uses no
|
|
# SQLite-specific syntax; this adapter only binds/decodes text values.
|
|
return Result(self.connection.execute(sql.replace('?', '%s'), tuple(map(text.encode, params))))
|
|
|
|
def next_id(self, table):
|
|
if table not in ('submissions', 'decisions'):
|
|
raise ValueError('Unsupported explicit identity sequence')
|
|
return self.connection.execute("SELECT nextval(pg_get_serial_sequence(%s,'id'))", (quote(self.schema) + '.' + quote(table),)).fetchone()[0]
|
|
|
|
def commit(self):
|
|
self.connection.commit()
|
|
|
|
def rollback(self):
|
|
self.connection.rollback()
|
|
|
|
def close(self):
|
|
# Ends the session and releases its advisory lock, including after an
|
|
# error/rollback. File/chunk commits deliberately retain the writer lock.
|
|
self.connection.close()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, kind, error, traceback):
|
|
self.rollback() if kind else self.commit()
|
|
|
|
|
|
class PostgresStore(Store):
|
|
def __init__(self, service, schema=None):
|
|
self.conn = WriteConnection(service, schema)
|
|
from paper_atlas.db.events import has_event_key
|
|
self.event_identity = has_event_key(self.conn)
|