Source code for squabble.rules.disallow_foreign_key

from pglast.enums import ConstrType

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

[docs]class DisallowForeignKey(BaseRule): """ Prevent creation of new ``FOREIGN KEY`` constraints. Optionally, can be configured with a list of table names that ARE allowed to create foreign key references. This rule will check ``CREATE TABLE`` and ``ALTER TABLE`` statements for foreign keys. Configuration :: { "DisallowForeignKey": { "excluded": ["table1", "table2"] } } """
[docs] class DisallowedForeignKeyConstraint(Message): """ Sometimes, foreign keys are not possible, or may cause more overhead than acceptable. If you're working with multiple services, each of which with their own database, it's not possible to create a foreign key reference to a table that exists on another database. In this case, you'll likely need to rely on your business logic being correct to guarantee referential integrity. A foreign key constraint requires the database to query the referenced table to ensure that the value exists. On high-traffic, write heavy production instances, this may cause unacceptable overhead on writes. """ TEMPLATE = '"{table}" has disallowed foreign key constraint' CODE = 1009
def enable(self, root_ctx, config): allowed_tables = set( name.lower() for name in config.get('excluded', []) ) # We want to check both columns that are part of CREATE TABLE # as well as ALTER TABLE ... ADD COLUMN root_ctx.register( 'CreateStmt', self._check_for_foreign_key(allowed_tables)) root_ctx.register( 'AlterTableStmt', self._check_for_foreign_key(allowed_tables)) @squabble.rule.node_visitor def _check_for_foreign_key(self, ctx, node, allowed_tables): """ Make sure ``node`` doesn't have a FOREIGN KEY reference. Coincidentally, both ``AlterTableStmt`` and ``CreateStmt`` have a similar enough structure that we can use the same function for both. """ table_name = node.relation.relname.value.lower() # No need to check further after this if table_name in allowed_tables: return def _check_constraint(child_ctx, constraint): if constraint.contype == ConstrType.CONSTR_FOREIGN: self.DisallowedForeignKeyConstraint(table=table_name), node=constraint ) ctx.register('Constraint', _check_constraint)