# -*- coding: utf-8 -*-
"""
The shared plumbing underneath every psycodict object.
:class:`PostgresBase` is the common base of the database, table and
statistics classes; it owns statement execution through ``_execute``
(logging, slow-query warnings, commit/rollback bookkeeping and
reconnection) together with helpers for inspecting tables, indexes and
constraints. The module also defines the layout of the ``meta_*`` tables --
the column lists, types and creation statements shared by everything that
reads or writes them -- and the metadata format version (``META_FORMAT``)
stamped into ``meta_format``.
"""
import csv
import logging
import re
import sys
import time
from collections import defaultdict
from psycopg import (
ClientCursor,
DatabaseError,
InterfaceError,
OperationalError,
ProgrammingError,
NotSupportedError,
DataError,
)
from psycopg.sql import SQL, Identifier, Placeholder, Literal, Composable
from .encoding import Json
from .utils import reraise, DelayCommit, QueryLogFilter
from .validation import (
InvalidDefinitionError,
check_new_table_name,
derived_identifier,
validate_constraint_definition,
validate_index_definition,
)
# What psycodict validates before putting it into a statement lives in one
# module. These names are re-exported here because base.py is where the rest
# of psycodict, and anything downstream, has always imported them from.
from .validation import ( # noqa: F401
InvalidColumnTypeError,
column_type_sql,
number_types,
param_types_whitelist,
types_whitelist,
validate_column_type,
)
##################################################################
# meta_* infrastructure #
##################################################################
[docs]
def jsonb_idx(cols, cols_type):
"""
The positions in ``cols`` whose type is ``jsonb``, as a tuple of
indexes. Used to decide which values need json decoding when reading
rows of the ``meta_*`` tables.
INPUT:
- ``cols`` -- a list of column names
- ``cols_type`` -- a dictionary mapping column names to their types
"""
return tuple(i for i, elt in enumerate(cols) if cols_type[elt] == "jsonb")
# The version of the metadata format described by the constants below: the
# layout of the meta_* tables, versioned by a single integer. That integer is
# a protocol revision of its own, not psycodict's major version -- a compatible
# additive revision may ship in a minor release, while one that raises
# min_compat past a supported client needs a major one. The format of a
# database is stamped into the single-row meta_format table as
# (version, min_compat), and every connection checks it: an older but
# compatible format connects with a warning and reduced functionality, while
# a layout this psycodict cannot safely use is refused. The policy, and the
# checklist to follow when changing the format, live in MetadataFormats.md.
#
# History:
# 0 -- the baseline (psycodict 0.x): meta_tables, meta_indexes,
# meta_constraints and their _hist counterparts, with no format stamp.
# An unstamped database that has meta tables is format 0.
# 1 -- (psycodict 1.0) meta_indexes/meta_indexes_hist gained a nullable
# ``whereclause`` column, holding the predicate of a partial index
# (NULL for an ordinary index). Compatible: against a format-0
# database everything keeps working except creating partial indexes.
# Migrate with ``PostgresDatabase.upgrade_metadata`` (or connect with
# upgrade=True).
META_FORMAT = 1
# The format of a search-data export file (the searchfile of copy_to/copy_from/
# reload), which is a *different* thing from the metadata format above: this
# describes the layout of a data file, not of the meta_* tables. A file may
# begin with a line ``# psycodict-export-format: N``; a file without one is
# format 0, the historical layout of just names, types and a blank line, so
# every file psycodict has ever written still reads. New files are written at
# EXPORT_FORMAT; a reader refuses a version it does not understand before
# loading any data. See Versioning.md for the compatibility promise.
EXPORT_FORMAT = 1
EXPORT_FORMAT_MARKER = "# psycodict-export-format:"
# The whole of the marker grammar: the prefix, then a nonnegative decimal
# version and nothing else. Only a line matching this is a version; a line
# that merely begins with the prefix may well be a format-0 row of column
# names, since a column may be called anything that is not empty and holds no
# control characters (see validate_column_name).
_EXPORT_FORMAT_MARKER_RE = re.compile(
r"%s[ \t]*([0-9]+)" % re.escape(EXPORT_FORMAT_MARKER)
)
_meta_tables_cols = (
"name",
"sort",
"count_cutoff",
"id_ordered",
"out_of_order",
"stats_valid",
"label_col",
"total",
"important",
"include_nones",
)
_meta_tables_cols_notrequired = (
"count_cutoff",
"stats_valid",
"total",
"important",
"include_nones",
)
# SQL literals giving the default values for the columns above
_meta_tables_defaults = {
"count_cutoff": "1000",
"stats_valid": "true",
"total": "0",
"important": "false",
"include_nones": "true",
}
_meta_tables_types = dict(zip(_meta_tables_cols, (
"text",
"jsonb",
"smallint",
"boolean",
"boolean",
"boolean",
"text",
"bigint",
"boolean",
"boolean",
)))
_meta_tables_jsonb_idx = jsonb_idx(_meta_tables_cols, _meta_tables_types)
_meta_indexes_cols = (
"index_name",
"table_name",
"type",
"columns",
"modifiers",
"storage_params",
# The predicate of a partial index (raw SQL), or NULL for an ordinary
# index. Added in metadata format 1; see META_FORMAT.
"whereclause",
)
_meta_indexes_types = dict(
zip(_meta_indexes_cols, ("text", "text", "text", "jsonb", "jsonb", "jsonb", "text"))
)
_meta_indexes_jsonb_idx = jsonb_idx(_meta_indexes_cols, _meta_indexes_types)
_meta_constraints_cols = (
"constraint_name",
"table_name",
"type",
"columns",
"check_func",
)
_meta_constraints_types = dict(
zip(_meta_constraints_cols, ("text", "text", "text", "jsonb", "text"))
)
_meta_constraints_jsonb_idx = jsonb_idx(_meta_constraints_cols, _meta_constraints_types)
# Columns introduced by a metadata format bump: column -> the format that
# added it; columns not listed are part of the format-0 baseline. A format
# bump must append its columns at the end of the _cols tuple above (see
# MetadataFormats.md), so that the columns of an older format are a prefix of
# the current ones.
_meta_col_formats = {
"meta_tables": {},
"meta_indexes": {"whereclause": 1},
"meta_constraints": {},
}
def _meta_cols_types_jsonb_idx(meta_name, fmt=None):
"""
The (columns, types, jsonb column indexes) of a metadata table.
``fmt`` restricts the columns to those present in that metadata format
(a prefix of the current ones, since format bumps only append columns);
the default is the current format. Callers touching a live database
should pass the connection's format, ``self._db._meta_format``, so that
their SQL matches the columns the database actually has.
"""
if meta_name not in ("meta_tables", "meta_indexes", "meta_constraints"):
raise ValueError("Unknown metadata table %r" % (meta_name,))
if meta_name == "meta_tables":
meta_cols = _meta_tables_cols
meta_types = _meta_tables_types
meta_jsonb_idx = _meta_tables_jsonb_idx
elif meta_name == "meta_indexes":
meta_cols = _meta_indexes_cols
meta_types = _meta_indexes_types
meta_jsonb_idx = _meta_indexes_jsonb_idx
elif meta_name == "meta_constraints":
meta_cols = _meta_constraints_cols
meta_types = _meta_constraints_types
meta_jsonb_idx = _meta_constraints_jsonb_idx
if fmt is not None and fmt < META_FORMAT:
added = _meta_col_formats[meta_name]
meta_cols = tuple(col for col in meta_cols if added.get(col, 0) <= fmt)
meta_jsonb_idx = jsonb_idx(meta_cols, meta_types)
return meta_cols, meta_types, meta_jsonb_idx
def _meta_table_name(meta_name):
meta_cols, _, _ = _meta_cols_types_jsonb_idx(meta_name)
# the column which will match search_table
table_name = "table_name"
if "name" in meta_cols:
table_name = "name"
return table_name
[docs]
class PostgresBase():
"""
A base class for various objects that interact with Postgres.
Any class inheriting from this one must provide a connection
to the postgres database, as well as a name used when creating a logger.
"""
def __init__(self, loggername, db):
# Have to record this object in the db so that we can reset the connection if necessary.
# This function also sets self.conn
db._register_object(self)
self._db = db
logging_options = db.config.options["logging"]
self.slow_cutoff = logging_options["slowcutoff"]
self._logger = l = logging.getLogger(loggername)
l.propagate = False
# we only want 2 handlers
l.handlers = []
l.setLevel(logging_options.get('loglevel', logging.INFO))
formatter = logging.Formatter("%(asctime)s - %(message)s")
fhandler = logging.FileHandler(logging_options["slowlogfile"])
fhandler.setFormatter(formatter)
fhandler.addFilter(QueryLogFilter())
l.addHandler(fhandler)
shandler = logging.StreamHandler()
shandler.setFormatter(formatter)
l.addHandler(shandler)
def _connection_reset(self):
"""
Refresh state this object derived from the database session.
Called on every registered object after a replacement connection has
been verified and adopted, together with the capability snapshot read
from it: by the time this runs, ``self.conn`` is the replacement and
``self._db``'s capability accessors describe it, so an override can
simply read them.
A no-op here. It exists for objects that cache something a new session
can invalidate -- whether the connected role may write to a particular
table, say, which changes if a reconnect reaches a standby or a role
whose grants have changed. An override that cannot confirm such a
permission should fail closed rather than leave the old answer in
place, and should use ``self.conn`` directly rather than ``_execute``,
which is the path being recovered from.
"""
def _mogrify(self, query, values):
"""
Render a query with values interpolated, for logging and error messages.
psycopg3 only supports client-side interpolation through ClientCursor,
so we create a temporary one (psycopg2 had mogrify on every cursor).
"""
return ClientCursor(self.conn).mogrify(query, values)
def _execute(
self,
query,
values=None,
silent=None,
values_list=False,
template=None,
commit=None,
slow_note=None,
reissued=False,
buffered=False
):
"""
Execute an SQL command, properly catching errors and returning the resulting cursor.
INPUT:
- ``query`` -- an SQL Composable object, the SQL command to execute.
- ``values`` -- values to substitute for %s in the query. Quoting from the documentation
for psycopg2 (https://initd.org/psycopg/docs/usage.html#passing-parameters-to-sql-queries):
Never, never, NEVER use Python string concatenation (+) or string parameters
interpolation (%) to pass variables to a SQL query string. Not even at gunpoint.
- ``silent`` -- boolean (default None). If True, don't log a warning for a slow query.
If None, allow DelayCommit contexts to control silencing.
- ``values_list`` -- boolean (default False). If True, use the ``execute_values`` method,
designed for inserting multiple values.
- ``template`` -- string, for use with ``values_list`` to insert constant values:
for example ``"(%s, %s, 42)"``. See the documentation of ``execute_values``
for more details.
- ``commit`` -- boolean (default None). Whether to commit changes on success. The default
is to commit unless we are currently in a DelayCommit context.
- ``slow_note`` -- a tuple for generating more useful data for slow query logging.
- ``reissued`` -- used internally to prevent infinite recursion when attempting to
reset the connection.
- ``buffered`` -- whether to create a server side cursor that must be manually
closed and connection committed (to closed the transaction) after using it,
this implies ``commit=False``.
.. NOTE:
If the Postgres connection has been closed, the execute statement will fail.
We try to recover gracefully by attempting to open a new connection
and issuing the command again. However, this approach is not prudent if this
execute statement is one of a chain of statements, which we detect by checking
whether ``commit == False``. In this case, we will reset the connection but reraise
the interface error.
The upshot is that you should use ``commit=False`` even for the last of a chain of
execute statements, then explicitly call ``self.conn.commit()`` afterward.
OUTPUT:
- a cursor object from which the resulting records can be obtained via iteration.
This function will also log slow queries.
"""
if not isinstance(query, Composable):
raise TypeError("You must use the psycopg.sql module to execute queries")
if buffered:
if commit is None:
commit = False
elif commit:
raise ValueError("buffered and commit are incompatible")
try:
cur = self._db._cursor(buffered=buffered)
t = time.time()
if values_list:
# This used to use psycopg2's execute_values; with psycopg3
# we expand the single "VALUES %s" placeholder to a per-row
# template and rely on executemany, which batches efficiently
# using pipeline mode.
if values:
if template is not None:
template = template.as_string(self.conn)
else:
template = "(" + ",".join(["%s"] * len(values[0])) + ")"
cur.executemany(query.as_string(self.conn).replace("%s", template, 1), values)
else:
try:
cur.execute(query, values)
except (OperationalError, ProgrammingError, NotSupportedError, DataError, SyntaxError) as e:
try:
context = " happens while executing {}".format(self._mogrify(query, values))
except Exception:
context = " happens while executing {} with values {}".format(query, values)
reraise(type(e), type(e)(str(e) + context), sys.exc_info()[2])
if silent is False or (silent is None and not self._db._silenced):
t = time.time() - t
if t > self.slow_cutoff:
if values_list:
query = query.as_string(self.conn).replace("%s", "VALUES_LIST")
elif values:
try:
query = self._mogrify(query, values)
except Exception:
# This shouldn't happen since the execution above was successful
query = query + str(values)
else:
query = query.as_string(self.conn)
if isinstance(query, bytes): # PY3 compatibility
query = query.decode("utf-8")
self._logger.info(query + " ran in \033[91m {0!s}s \033[0m".format(t))
if slow_note is not None:
self._logger.info(
"Replicate with db.%s.%s(%s)",
slow_note[0],
slow_note[1],
", ".join(str(c) for c in slow_note[2:]),
)
except (DatabaseError, InterfaceError):
if self.conn.closed != 0:
# If reissued, we need to raise since we're recursing.
if reissued:
raise
# Attempt to reset the connection
self._db.reset_connection()
if commit or (commit is None and self._db._nocommit_stack == 0):
return self._execute(
query,
values=values,
silent=silent,
values_list=values_list,
template=template,
commit=commit,
slow_note=slow_note,
buffered=buffered,
reissued=True,
)
else:
raise
else:
self.conn.rollback()
raise
else:
if commit or (commit is None and self._db._nocommit_stack == 0):
self.conn.commit()
return cur
def _table_exists(self, tablename):
"""
Check whether the specified table exists
INPUT:
- ``tablename`` -- a string, the name of the table
"""
cur = self._execute(
SQL("SELECT 1 FROM pg_tables WHERE schemaname = %s AND tablename = %s"),
[self._db.schema, tablename],
silent=True,
)
return cur.fetchone() is not None
def _all_tablenames(self):
"""
Return all (postgres) table names in the database
"""
return [
rec[0]
for rec in self._execute(
SQL("SELECT tablename FROM pg_tables WHERE schemaname = %s ORDER BY tablename"),
[self._db.schema],
silent=True,
)
]
def _get_locks(self):
return self._execute(SQL(
"SELECT t.relname, l.mode, l.pid, age(clock_timestamp(), a.backend_start) "
"FROM pg_locks l "
"JOIN pg_stat_all_tables t ON l.relation = t.relid JOIN pg_stat_activity a ON l.pid = a.pid "
"WHERE l.granted AND t.schemaname <> 'pg_toast'::name AND t.schemaname <> 'pg_catalog'::name"
))
def _table_locked(self, tablename, types="all"):
"""
Tests whether a table is locked.
INPUT:
- tablename -- a string, the name of the table
- types -- either a string describing the operation being performed
(which is translated to a list of lock types with which that operation conflicts)
or a list of lock types.
The valid strings are:
- 'update'
- 'delete'
- 'insert'
- 'index'
- 'select'
- 'all' (includes all locks)
The valid lock types to filter on are:
- 'AccessShareLock'
- 'RowShareLock'
- 'RowExclusiveLock'
- 'ShareUpdateExclusiveLock'
- 'ShareLock'
- 'ShareRowExclusiveLock'
- 'ExclusiveLock'
- 'AccessExclusiveLock'
OUTPUT:
A list of pairs (locktype, pid) where locktype is a string as above,
and pid is the process id of the postgres transaction holding the lock.
"""
if isinstance(types, str):
if types in ["update", "delete", "insert"]:
types = [
"ShareLock",
"ShareRowExclusiveLock",
"ExclusiveLock",
"AccessExclusiveLock",
]
elif types == "index":
types = [
"RowExclusiveLock",
"ShareUpdateExclusiveLock",
"ShareRowExclusiveLock",
"ExclusiveLock",
"AccessExclusiveLock",
]
elif types == "select":
types = [
"AccessExclusiveLock"
]
elif types != "all":
raise ValueError("Invalid lock type")
if types != "all":
good_types = [
"AccessShareLock",
"RowShareLock",
"RowExclusiveLock",
"ShareUpdateExclusiveLock",
"ShareLock",
"ShareRowExclusiveLock",
"ExclusiveLock",
"AccessExclusiveLock",
]
bad_types = [locktype for locktype in types if locktype not in good_types]
if bad_types:
raise ValueError("Invalid lock type(s): %s" % (", ".join(bad_types)))
return [
(locktype, pid)
for (name, locktype, pid, t) in self._get_locks()
if name == tablename and (types == "all" or locktype in types) and pid != self.conn.info.backend_pid
]
def _index_exists(self, indexname, tablename=None):
"""
Check whether the specified index exists
INPUT:
- ``indexname`` -- a string, the name of the index
- ``tablename`` -- (optional) a string
OUTPUT:
If ``tablename`` specified, returns a boolean. If not, returns
``False`` if there is no index with this name, or the corresponding tablename
as a string if there is.
"""
if tablename:
cur = self._execute(
SQL(
"SELECT 1 FROM pg_indexes "
"WHERE schemaname = %s AND indexname = %s AND tablename = %s"
),
[self._db.schema, indexname, tablename],
silent=True,
)
return cur.fetchone() is not None
else:
cur = self._execute(
SQL("SELECT tablename FROM pg_indexes WHERE schemaname = %s AND indexname = %s"),
[self._db.schema, indexname],
silent=True,
)
table = cur.fetchone()
if table is None:
return False
else:
return table[0]
def _relation_exists(self, name):
"""
Check whether the specified relation exists. Relations are indexes or constraints.
INPUT:
- ``name`` -- a string, the name of the relation
"""
cur = self._execute(
SQL(
"SELECT 1 FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
"WHERE n.nspname = %s AND c.relname = %s"
),
[self._db.schema, name],
)
return cur.fetchone() is not None
def _constraint_exists(self, constraintname, tablename=None):
"""
Check whether the specified constraint exists
INPUT:
- ``constraintname`` -- a string, the name of the index
- ``tablename`` -- (optional) a string
OUTPUT:
If ``tablename`` specified, returns a boolean. If not, returns
``False`` if there is no constraint with this name, or the corresponding tablename
as a string if there is.
"""
if tablename:
cur = self._execute(
SQL(
"SELECT 1 from information_schema.table_constraints "
"WHERE table_schema = %s AND table_name = %s AND constraint_name = %s"
),
[self._db.schema, tablename, constraintname],
silent=True,
)
return cur.fetchone() is not None
else:
cur = self._execute(
SQL(
"SELECT table_name from information_schema.table_constraints "
"WHERE table_schema = %s AND constraint_name = %s"
),
[self._db.schema, constraintname],
silent=True,
)
table = cur.fetchone()
if table is None:
return False
else:
return table[0]
def _list_indexes(self, tablename):
"""
Lists built index names on the search table ``tablename``
"""
cur = self._execute(
SQL("SELECT indexname FROM pg_indexes WHERE schemaname = %s AND tablename = %s"),
[self._db.schema, tablename],
silent=True,
)
return [elt[0] for elt in cur]
def _list_constraints(self, tablename):
"""
Lists constraint names on the search table ``tablename``
"""
# if we look into information_schema.table_constraints
# we also get internal constraints, I'm not sure why
# Alternatively, we do a triple join to get the right answer
cur = self._execute(
SQL(
"SELECT con.conname "
"FROM pg_catalog.pg_constraint con "
"INNER JOIN pg_catalog.pg_class rel "
" ON rel.oid = con.conrelid "
"INNER JOIN pg_catalog.pg_namespace nsp "
" ON nsp.oid = connamespace "
"WHERE nsp.nspname = %s AND rel.relname = %s"
),
[self._db.schema, tablename],
silent=True,
)
return [elt[0] for elt in cur]
def _rename_if_exists(self, name, suffix=""):
"""
Rename an index or constraint if it exists, appending ``_depN`` if so.
INPUT:
- ``name`` -- a string, the name of an index or constraint
- ``suffix`` -- a suffix to append to the name
"""
existing = derived_identifier(name, suffix)
if self._relation_exists(existing):
# First we determine its type
kind = None
tablename = self._constraint_exists(existing)
if tablename:
kind = "Constraint"
begin_renamer = SQL("ALTER TABLE {0} RENAME CONSTRAINT").format(Identifier(tablename))
end_renamer = SQL("{0} TO {1}")
begin_command = SQL("ALTER TABLE {0}").format(Identifier(tablename))
end_command = SQL("DROP CONSTRAINT {0}")
elif self._index_exists(existing):
kind = "Index"
begin_renamer = SQL("")
end_renamer = SQL("ALTER INDEX {0} RENAME TO {1}")
begin_command = SQL("")
end_command = SQL("DROP INDEX {0}")
else:
raise ValueError(
"Relation with name "
+ existing
+ " already exists. And it is not an index or a constraint"
)
# Find a new name for the existing index. derived_identifier keeps
# the whole suffix and cuts the base, so a name already at the byte
# limit does not lose the _depN that distinguishes it.
i = 0
deprecated_name = derived_identifier(name, "_dep0" + suffix)
while self._relation_exists(deprecated_name):
i += 1
deprecated_name = derived_identifier(name, "_dep%s%s" % (i, suffix))
self._execute(
begin_renamer + end_renamer.format(Identifier(existing), Identifier(deprecated_name))
)
command = begin_command + end_command.format(Identifier(deprecated_name))
logging.warning(
"{} with name {} ".format(kind, existing)
+ "already exists. "
+ "It has been renamed to {} ".format(deprecated_name)
+ "and it can be deleted with the following SQL command:\n"
+ command.as_string(self.conn)
)
def _check_restricted_suffix(self, name, kind="Index", skip_dep=False):
"""
Checks to ensure that the given name doesn't end with one
of the following restricted suffixes:
- ``_tmp``
- ``_pkey``
- ``_oldN``
- ``_depN``
INPUT:
- ``name`` -- string, the name of an index or constraint
- ``kind`` -- either ``"Index"`` or ``"Constraint"`` (only used for error msg)
- ``skip_dep`` -- if true, allow ``_depN`` as a suffix
"""
tests = [(r"_old[\d]+$", "_oldN"), (r"_tmp$", "_tmp"), ("_pkey$", "_pkey")]
if not skip_dep:
# _rename_if_exists appends "_dep<N>" (no trailing underscore), so
# the guard must be anchored the same way as its _oldN sibling; the
# stray trailing "_" here meant it never matched a real deprecated
# name and the check was dead.
tests.append((r"_dep[\d]+$", "_depN"))
for match, message in tests:
# re.search, not re.match: these patterns are $-anchored
# suffixes, and match() would only ever find them at the start
# of the name, so the guard never fired.
if re.search(match, name):
raise ValueError(
"{} name {} is invalid, ".format(kind, name)
+ "cannot end in {}, ".format(message)
+ "try specifying a different name"
)
@staticmethod
def _sort_str(sort_list):
"""
Constructs a psycopg.sql.Composable object describing a sort order
for Postgres from a list of columns.
INPUT:
- ``sort_list`` -- a list, either of strings (which are interpreted as
column names in the ascending direction) or of pairs (column name, 1 or -1).
OUTPUT:
- a Composable to be used by psycopg in the ORDER BY clause.
"""
PostgresBase._check_sort_duplicates(sort_list)
L = []
for col in sort_list:
if isinstance(col, str):
L.append(Identifier(col))
elif col[1] == 1:
L.append(Identifier(col[0]))
else:
L.append(SQL("{0} DESC NULLS LAST").format(Identifier(col[0])))
return SQL(", ").join(L)
@staticmethod
def _check_sort_duplicates(sort_list):
"""
Raise if a column appears more than once in ``sort_list`` (a list of
column names or (column, direction) pairs). A column already fixes the
order by its first appearance, so a repeat is dead weight and almost
always a mistake.
"""
seen = set()
for col in sort_list:
name = col if isinstance(col, str) else col[0]
if name in seen:
raise ValueError("Duplicate column %r in sort order" % (name,))
seen.add(name)
def _column_types(self, table_name, data_types=None):
"""
Returns the
- column list,
- column types (as a dict), and
- has_id for a given table_name or list of table names
INPUT:
- ``table_name`` -- a string or list of strings
- ``data_types`` -- (optional) a dictionary providing a list of column names and
types for each table name. If not provided, will be looked up from the database.
EXAMPLES::
>>> db._column_types('nonexistent')
([], {}, False)
>>> db._column_types('test_fields')
(['class_group', 'class_number', 'degree', 'disc_abs',
'disc_sign', 'label', 'r2', 'ramps'],
{'id': 'bigint',
'class_number': 'integer',
'disc_abs': 'integer',
'degree': 'smallint',
'disc_sign': 'smallint',
'r2': 'smallint',
'ramps': 'integer[]',
'class_group': 'jsonb',
'label': 'text'},
True)
"""
has_id = False
col_list = []
col_type = {}
if isinstance(table_name, str):
table_name = [table_name]
for tname in table_name:
if data_types is None or tname not in data_types:
# in case of an array data type, data_type only gives 'ARRAY', while 'udt_name::regtype' gives us 'base_type[]'
cur = self._execute(
SQL(
"SELECT column_name, udt_name::regtype FROM information_schema.columns "
"WHERE table_schema = %s AND table_name = %s ORDER BY ordinal_position"
),
[self._db.schema, tname],
)
else:
cur = data_types[tname]
for rec in cur:
col = rec[0]
if col in col_type and col_type[col] != rec[1]:
raise ValueError("Type mismatch on %s: %s vs %s" % (col, col_type[col], rec[1]))
col_type[col] = rec[1]
if col != "id":
col_list.append(col)
else:
has_id = True
return sorted(col_list), col_type, has_id
def _relation_columns(self, table):
"""
The set of column names of ``table``, or None if it has none.
Used to check an index or constraint definition against the relation it
will be built on at the moment it is built. None -- for a relation
that does not exist yet, such as the ``_tmp`` table of a reload that has
not created it -- means "unknown", and leaves the columns unchecked
here so that PostgreSQL gives its own error rather than a misleading
one about columns.
"""
cur = self._execute(
SQL(
"SELECT column_name FROM information_schema.columns "
"WHERE table_schema = %s AND table_name = %s"
),
[self._db.schema, table],
silent=True,
commit=False,
)
columns = {rec[0] for rec in cur}
return columns or None
def _copy_to_select(self, select, filename, header="", sep="|", silent=False):
"""
Using COPY ... TO STDOUT, exports the data from a select statement.
INPUT:
- ``select`` -- an SQL Composable object giving a select statement
- ``header`` -- An initial header to write to the file
- ``sep`` -- a separator, defaults to ``|``
- ``silent`` -- suppress reporting success
"""
if sep != "\t":
sep_clause = SQL(" (DELIMITER {0})").format(Literal(sep))
else:
sep_clause = SQL("")
copyto = SQL("COPY ({0}) TO STDOUT{1}").format(select, sep_clause)
with open(filename, "w") as F:
try:
F.write(header)
cur = self._db._cursor()
with cur.copy(copyto) as copy:
for data in copy:
F.write(bytes(data).decode())
except Exception:
self.conn.rollback()
raise
else:
if not silent:
print("Created file %s" % filename)
def _check_header_lines(
self, F, table_name, columns_set, sep="|", prohibit_missing=True
):
"""
Reads the header lines from a file (an optional format marker, the row
of column names, the row of column types and the blank line), checking
if these names match the columns set and the types match the expected
types in the table.
Returns a list of column names present in the header.
INPUT:
- ``F`` -- an open file handle, at the beginning of the file.
- ``table_name`` -- the table to compare types against (or a list of tables)
- ``columns_set`` -- a set of the columns expected in the table.
- ``sep`` -- a string giving the column separator.
- ``prohibit_missing`` -- raise an error if not all columns present.
OUTPUT:
The ordered list of columns. The first entry may be ``"id"`` if the data
contains an id column.
"""
col_list, col_type, _ = self._column_types(table_name)
columns_set.discard("id")
if not (columns_set <= set(col_list)):
raise ValueError("{} is not a subset of {}".format(columns_set, col_list))
header_cols = self._read_header_lines(F, sep=sep)
names = [elt[0] for elt in header_cols]
names_set = set(names)
if "id" in names_set:
if names[0] != "id":
raise ValueError("id must be the first column")
if header_cols[0][1] not in ["int2", "smallint", "int4", "integer", "int8", "bigint"]:
raise ValueError("id must be of integeral type")
names_set.discard("id")
header_cols = header_cols[1:]
missing = columns_set - names_set
extra = names_set - columns_set
wrong_type = [
(name, typ)
for name, typ in header_cols
if name in columns_set and col_type[name] != typ
]
if (missing and prohibit_missing) or extra or wrong_type:
err = ""
if missing or extra:
err += "Invalid header: "
if missing:
err += ", ".join(list(missing)) + " (missing)"
if extra:
err += ", ".join(list(extra)) + " (extra)"
if wrong_type:
if len(wrong_type) > 1:
err += "Invalid types: "
else:
err += "Invalid type: "
err += ", ".join(
"%s should be %s instead of %s" % (name, col_type[name], typ)
for name, typ in wrong_type
)
raise ValueError(err)
return names
def _copy_from_stdin(self, F, table, columns=None, sep=None, null=r"\N"):
"""
Stream an open file object into a table using COPY ... FROM STDIN.
This replaces psycopg2's ``cursor.copy_from``, which was removed in
psycopg3 in favor of an explicit COPY statement. Returns the cursor,
whose ``rowcount`` gives the number of rows loaded.
INPUT:
- ``F`` -- an open file object to read from
- ``table`` -- the name of the table to load into
- ``columns`` -- the columns present in the file, in order
(defaults to all columns in table order)
- ``sep`` -- the column separator (defaults to postgres' text-format
default, a tab, like psycopg2's copy_from did)
- ``null`` -- the null marker (the text-format default)
"""
if columns is None:
cols = SQL("")
else:
cols = SQL(" ({0})").format(SQL(", ").join(map(Identifier, columns)))
if sep is None:
options = SQL("")
else:
options = SQL(" WITH (DELIMITER {0}, NULL {1})").format(Literal(sep), Literal(null))
copy_sql = SQL("COPY {0}{1} FROM STDIN{2}").format(Identifier(table), cols, options)
cur = self._db._cursor()
with cur.copy(copy_sql) as copy:
while True:
chunk = F.read(1 << 20)
if not chunk:
break
copy.write(chunk)
return cur
def _copy_from(self, filename, table, columns, header, kwds):
"""
Helper function for ``copy_from`` and ``reload``.
INPUT:
- ``filename`` -- the filename to load
- ``table`` -- the table into which the data should be added
- ``columns`` -- a list of columns to load (the file may contain them in
a different order, specified by a header row)
- ``header`` -- whether the file has header rows ordering the columns.
This should be True for search tables, False for counts and stats.
- ``kwds`` -- may contain ``sep`` and ``null`` options for the COPY
"""
kwds = dict(kwds) # to not modify the dict kwds, with the pop
sep = kwds.pop("sep", "|")
null = kwds.pop("null", r"\N")
kwds.pop("size", None) # psycopg2 buffer size, no longer meaningful
if kwds:
raise TypeError("Unsupported copy_from options: %s" % ", ".join(kwds))
with DelayCommit(self, silence=True):
with open(filename) as F:
if header:
# This consumes the header, leaving F at the first data row
columns = self._check_header_lines(F, table, set(columns), sep=sep)
addid = "id" not in columns
else:
addid = False
if addid:
# create sequence
# The values are inlined as literals: DDL statements
# cannot take parameters under psycopg3's server-side
# binding (psycopg2 interpolated them client-side).
cur_count = self.max_id(table)
seq_name = table + "_seq"
create_seq = SQL(
"CREATE SEQUENCE {0} START WITH {1} MINVALUE {1} CACHE 10000"
).format(Identifier(seq_name), Literal(cur_count + 1))
self._execute(create_seq)
# edit default value
alter_table = SQL(
"ALTER TABLE {0} ALTER COLUMN {1} SET DEFAULT nextval({2})"
).format(Identifier(table), Identifier("id"), Literal(seq_name))
self._execute(alter_table)
cur = self._copy_from_stdin(F, table, columns, sep, null=null)
if addid:
alter_table = SQL(
"ALTER TABLE {0} ALTER COLUMN {1} DROP DEFAULT"
).format(Identifier(table), Identifier("id"))
self._execute(alter_table)
drop_seq = SQL("DROP SEQUENCE {0}").format(Identifier(seq_name))
self._execute(drop_seq)
return addid, cur.rowcount
def _get_tablespace(self):
# overridden in table and statstable
pass
def _tablespace_clause(self, tablespace=None):
"""
A clause for use in CREATE statements
"""
if tablespace is None:
tablespace = self._get_tablespace()
if tablespace is None:
return SQL("")
else:
return SQL(" TABLESPACE {0}").format(Identifier(tablespace))
def _clone(self, table, tmp_table):
"""
Utility function: creates a table with the same schema as the given one.
INPUT:
- ``table`` -- string, the name of an existing table
- ``tmp_table`` -- string, the name of the new table to create
Every relation psycodict swaps through -- ``_tmp``, and by way of the
swap ``_oldN`` -- is created here, so this is where a name PostgreSQL
would truncate into another relation's is refused.
"""
check_new_table_name(tmp_table)
if self._table_exists(tmp_table):
# remove suffix for display message
for suffix in ['_counts', '_stats']:
if table.endswith(suffix):
table = table[:-len(suffix)]
raise ValueError(
"Temporary table %s already exists. "
"Run db.%s.cleanup_from_reload() if you want to delete it and proceed."
% (tmp_table, table)
)
# A bare LIKE copies only the column names and types; carry over the
# per-column STORAGE settings (and COMPRESSION, once the server knows
# about it) so that clones -- and hence reload and staged, which swap
# a clone into place -- do not silently reset them to the defaults.
including = SQL(" INCLUDING STORAGE")
version = int(self._execute(
SQL("SELECT current_setting('server_version_num')"), silent=True
).fetchone()[0])
if version >= 140000:
# INCLUDING COMPRESSION appeared in PostgreSQL 14 together with
# per-column compression itself
including += SQL(" INCLUDING COMPRESSION")
creator = SQL("CREATE TABLE {0} (LIKE {1}{2}){3}").format(Identifier(tmp_table), Identifier(table), including, self._tablespace_clause())
self._execute(creator)
def _check_col_datatype(self, typ):
"""
The spelling of the column type ``typ`` to use in DDL, or ``ValueError``.
A thin method wrapper around :func:`validate_column_type`; callers must
build their SQL from the returned spelling rather than from ``typ``.
"""
spelling, _ = validate_column_type(typ)
return spelling
def _pairs_to_dict(self, L):
"""
Standardize input format for search_columns
"""
if L is None:
return L
D = defaultdict(list)
for (col, typ) in L:
D[typ].append(col)
return D
def _get_type_sortkey(self, typ):
"""
Returns the negated storage cost, together with the type
Used to sort columns when creating a table for smaller storage footprint
"""
spelling, cost = validate_column_type(typ)
return -cost, spelling
def _order_columns(self, coldict, addid="bigint"):
"""
For space reasons, we sort the columns by type, then alphabetically within each type
This function returns the correct order of the columns.
coldict should be in the format output by _pairs_to_dict.
"""
if addid and not any("id" in vals for vals in coldict.values()):
if addid not in coldict: # coldict might be a normal dictionary, not a defaultdict
coldict[addid] = []
coldict[addid].append("id")
allcols = []
# Validate every type before any of them reaches the statement, so that
# an invalid one raises rather than being interpolated: the type has to
# go in as SQL text (PostgreSQL has no placeholder for a type), and only
# the spelling the validator returns is safe to emit.
validated = {typ: validate_column_type(typ) for typ in coldict}
dictorder = sorted(coldict, key=lambda typ: (-validated[typ][1], validated[typ][0]))
for typ in dictorder:
for col in sorted(coldict[typ]):
allcols.append(SQL("{0} {1}").format(Identifier(col), column_type_sql(typ)))
return allcols
def _create_table(self, name, columns, addid="bigint", tablespace=None):
"""
Utility function: creates a table with the schema specified by ``columns``.
If self is a table, the new table will be in the same tablespace.
INPUT:
- ``name`` -- the desired name
- ``columns`` -- list of pairs, where the first entry is
the column name and the second one is the corresponding type
"""
# Defense in depth for every caller, including header-driven creation
# and internal copies: a relation is not created under a name
# PostgreSQL would truncate into another one's.
check_new_table_name(name)
if not isinstance(columns, dict):
columns = self._pairs_to_dict(columns)
ordered = self._order_columns(columns, addid=addid)
table_col = SQL(", ").join(self._order_columns(columns, addid=addid))
creator = SQL("CREATE TABLE {0} ({1}){2}").format(Identifier(name), table_col, self._tablespace_clause(tablespace))
self._execute(creator)
def _create_table_from_header(self, filename, name, sep, addid="bigint", tablespace=None):
"""
Utility function: creates a table with the schema specified in the header of the file.
Returns column names found in the header
INPUT:
- ``filename`` -- a string, the filename to load the table from
- ``name`` -- the name of the table
- ``sep`` -- the separator character, defaulting to tab
- ``addid`` -- if true, also adds an id column to the created table with the given type
OUTPUT:
The list of column names and types found in the header
"""
if self._table_exists(name):
error_msg = "Table %s already exists." % name
if name.endswith("_tmp"):
error_msg += (
"Run db.%s.cleanup_from_reload() "
"if you want to delete it and proceed." % (name[:-4])
)
raise ValueError(error_msg)
with open(filename, "r") as F:
columns = self._read_header_lines(F, sep)
col_list = [elt[0] for elt in columns]
self._create_table(name, columns, addid=addid, tablespace=tablespace)
return col_list
def _swap(self, tables, source, target):
"""
Renames tables, indexes, constraints and primary keys, for use in reload.
INPUT:
- ``tables`` -- a list of table names to reload (including suffixes like
``_extra`` or ``_counts`` but not ``_tmp``).
- ``source`` -- the source suffix for the swap.
- ``target`` -- the target suffix for the swap.
"""
rename_table = SQL("ALTER TABLE {0} RENAME TO {1}")
rename_constraint = SQL("ALTER TABLE {0} RENAME CONSTRAINT {1} TO {2}")
rename_index = SQL("ALTER INDEX {0} RENAME TO {1}")
def target_name(name, tablename, kind):
original_name = name[:]
if source != "" and name.endswith(source):
# drop the suffix
original_name = original_name[: -len(source)]
assert original_name + source == name
elif source != "":
logging.warning(
"{} of {} with name {}".format(kind, tablename, name)
+ " does not end with the suffix {}".format(source)
)
target_name = original_name + target
try:
self._check_restricted_suffix(original_name, kind, skip_dep=True)
except ValueError:
logging.warning(
"{} of {} with name {}".format(kind, tablename, name)
+ " uses a restricted suffix. "
+ "The name will be extended with a _ in the swap"
)
target_name = original_name + "_" + target
return target_name
# Every destination is decided and checked before the first rename.
# PostgreSQL truncates a name over its limit rather than refusing it, so
# a batch that would produce two identical destinations -- or one that
# collides with a relation already there -- has to stop before it has
# renamed half of them. The *sources* are not checked: they may be the
# concatenated spelling of a legacy relation the server truncated when
# it was created, which still addresses it.
renames = [(table + source, table + target) for table in tables]
for _, destination in renames:
check_new_table_name(destination)
destinations = [destination for _, destination in renames]
if len(set(destinations)) != len(destinations):
raise InvalidDefinitionError(
"This swap would rename two relations to the same name: %s"
% ", ".join(sorted(destinations))
)
with DelayCommit(self, silence=True):
for tablename_old, tablename_new in renames:
self._execute(rename_table.format(Identifier(tablename_old), Identifier(tablename_new)))
done = set() # done constraints/indexes
# We threat pkey separately
pkey_old = derived_identifier(tablename_old, "_pkey")
pkey_new = derived_identifier(tablename_new, "_pkey")
if self._constraint_exists(pkey_old, tablename_new):
self._execute(
rename_constraint.format(
Identifier(tablename_new),
Identifier(pkey_old),
Identifier(pkey_new),
)
)
done.add(pkey_new)
for constraint in self._list_constraints(tablename_new):
if constraint in done:
continue
c_target = target_name(constraint, tablename_new, "Constraint")
if c_target != constraint:
self._rename_if_exists(c_target)
self._execute(
rename_constraint.format(
Identifier(tablename_new),
Identifier(constraint),
Identifier(c_target),
)
)
done.add(c_target)
for index in self._list_indexes(tablename_new):
if index in done:
continue
i_target = target_name(index, tablename_new, "Index")
if i_target != index:
self._rename_if_exists(i_target)
self._execute(
rename_index.format(Identifier(index), Identifier(i_target))
)
done.add(i_target) # not really needed
def _read_header_lines(self, F, sep="|"):
"""
Reads the header lines from a search-data file (an optional format
marker, the row of column names, the row of column types, and the blank
line), returning the columns and their types.
INPUT:
- ``F`` -- an open file handle, at the beginning of the file.
- ``sep`` -- a string giving the column separator.
OUTPUT:
A list of pairs where the first entry is the column and the second the
corresponding type
A file may begin with ``# psycodict-export-format: N``. A file without
one is format 0, the historical layout, so every file psycodict has
ever written still reads. A version newer than this psycodict
understands is refused here, before any of the data is loaded.
A format-0 names row may itself begin with the marker prefix, since a
column may be called anything printable, so the prefix alone does not
decide which layout this is. The blank line does: a format-0 header is
``names / types / blank`` and a marked one ``marker / names / types /
blank``, so if the third physical line is blank the first line was a
row of column names.
"""
first = F.readline()
if first.strip().startswith(EXPORT_FORMAT_MARKER):
second = F.readline()
third = F.readline()
if third.strip():
# Not a format-0 header, so the first line really is a marker.
version = export_format_version(first)
if version > EXPORT_FORMAT:
raise ValueError(
"This file is psycodict export format %s, but this psycodict "
"understands only up to format %s; upgrade psycodict to read "
"it" % (version, EXPORT_FORMAT)
)
names_line, types_line = second, third
blank = F.readline()
else:
# Format 0, whose one column happens to be named like a marker.
names_line, types_line, blank = first, second, third
else:
# No marker: format 0, and this first line is the column names.
names_line = first
types_line = F.readline()
blank = F.readline()
names = [x.strip() for x in names_line.strip().split(sep)]
types = [x.strip() for x in types_line.strip().split(sep)]
if blank.strip():
raise ValueError("The header must end with a blank line")
if len(names) != len(types):
raise ValueError(
"The first line specifies %s columns, while the second specifies %s"
% (len(names), len(types))
)
return list(zip(names, types))
def _count_data_rows(self, filename, sep="|"):
"""
The number of data rows in a search-data file.
INPUT:
- ``filename`` -- the search-data file to count.
- ``sep`` -- a string giving the column separator, since the header is
parsed to find where the data starts.
OUTPUT:
The number of lines after the header. Counting physical lines instead
would count the header too, and a marked file's header is one line
longer than an unmarked one's, so the two formats would not agree on
what "1000 rows" means.
"""
with open(filename) as F:
self._read_header_lines(F, sep=sep)
return sum(1 for _ in F)
##################################################################
# Exporting, importing, reloading and reverting meta_* #
##################################################################
def _copy_to_meta(self, meta_name, filename, search_table, sep="|"):
# The columns this database actually has: an export from an
# older-format database carries that format's columns (a prefix of
# the current ones), which _meta_file_columns recognizes on import.
meta_cols, _, _ = _meta_cols_types_jsonb_idx(meta_name, self._db._meta_format)
table_name = _meta_table_name(meta_name)
table_name_sql = Identifier(table_name)
meta_name_sql = Identifier(meta_name)
cols_sql = SQL(", ").join(map(Identifier, meta_cols))
select = SQL("SELECT {} FROM {} WHERE {} = {}").format(
cols_sql, meta_name_sql, table_name_sql, Literal(search_table)
)
now = time.time()
with DelayCommit(self):
self._copy_to_select(select, filename, sep=sep, silent=True)
print(
"Exported %s for %s in %.3f secs"
% (meta_name, search_table, time.time() - now)
)
def _meta_file_columns(self, meta_name, filename, sep="|"):
"""
The columns of ``meta_name`` that an exported metadata file carries.
Metadata files have no header line, so the format they were exported
at is recovered from their width: format bumps only append columns,
so a file written at format f holds the first ``len(columns at f)``
of the current columns. Returns that column prefix, or None for an
empty file. A file wider than this database's meta table (exported
from a newer format than the database is at) or of a width matching
no known format is rejected here, with instructions, rather than
passed on to COPY to fail cryptically.
"""
with open(filename) as F:
first = next(csv.reader(F, delimiter=str(sep)), None)
if first is None:
return None
width = len(first)
db_cols, _, _ = _meta_cols_types_jsonb_idx(meta_name, self._db._meta_format)
# width -> the oldest format with that many columns
widths = {}
for fmt in range(META_FORMAT + 1):
widths.setdefault(len(_meta_cols_types_jsonb_idx(meta_name, fmt)[0]), fmt)
if width not in widths:
raise ValueError(
"The file %s has %s columns, which matches no known format of "
"%s (expected %s)"
% (filename, width, meta_name,
" or ".join(str(w) for w in sorted(widths)))
)
if width > len(db_cols):
raise ValueError(
"The file %s was exported from a database using metadata "
"format %s, but this database uses the older format %s: "
"migrate it with upgrade_metadata() (or reconnect with "
"upgrade=True) before reloading, or re-export the file from "
"a format-%s database."
% (filename, widths[width], self._db._meta_format,
self._db._meta_format)
)
return db_cols[:width]
def _validate_meta_rows(self, meta_name, meta_cols, rows, source):
"""
Check index or constraint definitions that have just been loaded.
INPUT:
- ``meta_name`` -- ``"meta_indexes"``, ``"meta_constraints"`` or
``"meta_tables"``
- ``meta_cols`` -- the columns the rows carry, in order
- ``rows`` -- the rows, as returned by the database (jsonb columns
already decoded)
- ``source`` -- where they came from, for the error message
Must be called inside the transaction that loaded the rows, so that
raising leaves neither the new definitions nor the deletion of the old
ones behind.
The columns of the definitions are not checked against the table here:
an index may legitimately name a column that a reload is about to add,
and the relation the definition will be built on need not exist yet.
Column existence is checked when the definition becomes DDL.
"""
if meta_name not in ("meta_indexes", "meta_constraints"):
return
for row in rows:
record = dict(zip(meta_cols, row))
try:
if meta_name == "meta_indexes":
validate_index_definition(
record["index_name"],
record["table_name"],
record["type"],
record["columns"],
record["modifiers"],
record["storage_params"],
record.get("whereclause"),
)
else:
validate_constraint_definition(
record["constraint_name"],
record["table_name"],
record["type"],
record["columns"],
record["check_func"],
# a PostgresTable attribute: metadata for a search
# table is always reloaded through its table object
valid_check_functions=getattr(self, "_valid_check_functions", ()),
)
except ValueError as err:
raise InvalidDefinitionError(
"%s in %s is not a definition psycodict can build: %s"
% (
record.get("index_name") or record.get("constraint_name"),
source,
err,
)
)
def _get_current_meta_version(self, meta_name, search_table):
# the column which will match search_table
table_name = _meta_table_name(meta_name)
table_name_sql = Identifier(table_name)
meta_name_hist_sql = Identifier(meta_name + "_hist")
res = self._execute(
SQL("SELECT MAX(version) FROM {} WHERE {} = %s").format(
meta_name_hist_sql, table_name_sql
),
[search_table],
).fetchone()[0]
if res is None:
res = -1
return res
def _reload_meta(self, meta_name, filename, search_table, sep="|"):
# The database's columns for the SELECT/INSERT below; the file may
# carry fewer (it was exported from an older format), in which case
# the trailing columns load as NULL.
meta_cols, _, jsonb_idx = _meta_cols_types_jsonb_idx(meta_name, self._db._meta_format)
file_cols = self._meta_file_columns(meta_name, filename, sep)
# the column which will match search_table
table_name = _meta_table_name(meta_name)
table_name_idx = meta_cols.index(table_name)
table_name_sql = Identifier(table_name)
meta_name_sql = Identifier(meta_name)
meta_name_hist_sql = Identifier(meta_name + "_hist")
with open(filename, "r") as F:
lines = list(csv.reader(F, delimiter=str(sep)))
if not lines:
return
for line in lines:
if line[table_name_idx] != search_table:
raise RuntimeError(
f"column {table_name_idx} (= {line[table_name_idx]}) "
f"in the file {filename} doesn't match "
f"the search table name {search_table}"
)
with DelayCommit(self, silence=True):
# delete the current columns
self._execute(
SQL("DELETE FROM {} WHERE {} = %s").format(meta_name_sql, table_name_sql),
[search_table],
)
# insert new columns
with open(filename, "r") as F:
try:
self._copy_from_stdin(F, meta_name, file_cols, sep)
except Exception:
self.conn.rollback()
raise
version = self._get_current_meta_version(meta_name, search_table) + 1
# copy the new rows to history
cols_sql = SQL(", ").join(map(Identifier, meta_cols))
rows = self._execute(
SQL("SELECT {} FROM {} WHERE {} = %s").format(cols_sql, meta_name_sql, table_name_sql),
[search_table],
)
cols = meta_cols + ("version",)
cols_sql = SQL(", ").join(map(Identifier, cols))
place_holder = SQL(", ").join(Placeholder() * len(cols))
query = SQL("INSERT INTO {} ({}) VALUES ({})").format(meta_name_hist_sql, cols_sql, place_holder)
imported = []
for row in rows:
imported.append(row)
row = [
Json(elt) if i in jsonb_idx else elt for i, elt in enumerate(row)
]
self._execute(query, row + [version])
# Validate what was imported, inside the transaction: a file that
# carries a definition psycodict would not build raises here, and
# the surrounding DelayCommit rolls back both the DELETE above and
# the rows just loaded, leaving the old metadata in place.
self._validate_meta_rows(meta_name, meta_cols, imported, filename)
def _revert_meta(self, meta_name, search_table, version=None):
meta_cols, _, jsonb_idx = _meta_cols_types_jsonb_idx(meta_name, self._db._meta_format)
# the column which will match search_table
table_name = _meta_table_name(meta_name)
table_name_sql = Identifier(table_name)
meta_name_sql = Identifier(meta_name)
meta_name_hist_sql = Identifier(meta_name + "_hist")
# by the default goes back one step
currentversion = self._get_current_meta_version(meta_name, search_table)
if currentversion == -1:
raise RuntimeError("No history to revert")
if version is None:
version = max(0, currentversion - 1)
with DelayCommit(self, silence=True):
# delete current rows
self._execute(
SQL("DELETE FROM {} WHERE {} = %s").format(meta_name_sql, table_name_sql),
[search_table],
)
# copy data from history
cols_sql = SQL(", ").join(map(Identifier, meta_cols))
rows = self._execute(
SQL("SELECT {} FROM {} WHERE {} = %s AND version = %s").format(
cols_sql, meta_name_hist_sql, table_name_sql
),
[search_table, version],
)
place_holder = SQL(", ").join(Placeholder() * len(meta_cols))
query = SQL("INSERT INTO {} ({}) VALUES ({})").format(meta_name_sql, cols_sql, place_holder)
cols = meta_cols + ("version",)
cols_sql = SQL(", ").join(map(Identifier, cols))
place_holder = SQL(", ").join(Placeholder() * len(cols))
query_hist = SQL("INSERT INTO {} ({}) VALUES ({})").format(
meta_name_hist_sql, cols_sql, place_holder
)
restored = []
for row in rows:
restored.append(row)
row = [Json(elt) if i in jsonb_idx else elt for i, elt in enumerate(row)]
self._execute(query, row)
self._execute(query_hist, row + [currentversion + 1])
# History is as untrusted as a file: the rows in it were written by
# whatever psycodict version was running at the time, and can be
# edited in place like any other table. Validating here, inside
# the DelayCommit, means a poisoned version cannot be reverted to.
self._validate_meta_rows(
meta_name, meta_cols, restored,
"%s_hist version %s" % (meta_name, version),
)