Source code for squabble.rules.add_column_disallow_constraints

import pglast
from pglast.enums import AlterTableType, ConstrType

import squabble.rule
from squabble import RuleConfigurationException
from squabble.message import Message
from squabble.rules import BaseRule


[docs]class AddColumnDisallowConstraints(BaseRule): """ Prevent adding a column with certain constraints to an existing table. Configuration: :: { "AddColumnDisallowConstraints": { "disallowed": ["DEFAULT", "FOREIGN"] } } Valid constraint types: - DEFAULT - NULL - NOT NULL - FOREIGN - UNIQUE """ _CONSTRAINT_MAP = { 'DEFAULT': ConstrType.CONSTR_DEFAULT, 'NULL': ConstrType.CONSTR_NULL, 'NOT NULL': ConstrType.CONSTR_NOTNULL, 'FOREIGN': ConstrType.CONSTR_FOREIGN, 'UNIQUE': ConstrType.CONSTR_UNIQUE, }
[docs] class ConstraintNotAllowed(Message): """ When adding a column to an existing table, certain constraints can have unintentional side effects, like locking the table or introducing performance issues. For example, adding a ``DEFAULT`` constraint may hold a lock on the table while all existing rows are modified to fill in the default value. A ``UNIQUE`` constraint will require scanning the table to confirm there are no duplicates. On a particularly hot table, a ``FOREIGN`` constraint will introduce possibly dangerous overhead to confirm the referential integrity of each row. """ CODE = 1004 TEMPLATE = 'column "{col}" has a disallowed constraint'
def enable(self, ctx, config): disallowed = config.get('disallowed', []) if disallowed == []: raise RuleConfigurationException( self, 'must specify `disallowed` constraints') constraints = set() for c in disallowed: ty = self._CONSTRAINT_MAP.get(c.upper()) if ty is None: raise RuleConfigurationException( self, 'unknown constraint: `%s`' % c) constraints.add(ty) ctx.register('AlterTableCmd', self._check(constraints)) @squabble.rule.node_visitor def _check(self, ctx, node, disallowed_constraints): """ Node is an `AlterTableCmd`: :: { 'AlterTableCmd': { 'def': { 'ColumnDef': { 'colname': 'bar', 'constraints': [{'Constraint': {'contype': 2}}] } } } } """ # We only care about adding a column if node.subtype != AlterTableType.AT_AddColumn: return constraints = node['def'].constraints # No constraints imposed, nothing to do. if constraints == pglast.Missing: return for constraint in constraints: if constraint.contype.value in disallowed_constraints: col = node['def'].colname.value ctx.report( self.ConstraintNotAllowed(col=col), node=constraint)