summaryrefslogtreecommitdiff
path: root/test/dialect/postgresql/test_reflection.py
diff options
context:
space:
mode:
authorcheremnov <32135863+cheremnov@users.noreply.github.com>2022-02-24 02:22:33 -0500
committerFederico Caselli <cfederico87@gmail.com>2022-06-29 09:13:37 +0000
commit5fb63bc1423e75812a24e809d16731a3282c2a12 (patch)
tree2e3293890e1b326146ea8848ceac9a65fae9490b /test/dialect/postgresql/test_reflection.py
parent6a560cf03c302d2ebd9ae7c7dc4d587983096ba4 (diff)
downloadsqlalchemy-5fb63bc1423e75812a24e809d16731a3282c2a12.tar.gz
Comments on (named) constraints
Adds support for comments on named constraints, including `ForeignKeyConstraint`, `PrimaryKeyConstraint`, `CheckConstraint`, `UniqueConstraint`, solving the [Issue 5667](https://github.com/sqlalchemy/sqlalchemy/issues/5667). Supports only PostgreSQL backend. ### Description Following the example of [Issue 1546](https://github.com/sqlalchemy/sqlalchemy/issues/1546), supports comments on constraints. Specifically, enables comments on _named_ ones — as I get it, PostgreSQL prohibits comments on unnamed constraints. Enables setting the comments for named constraints like this: ``` Table( 'example', metadata, Column('id', Integer), Column('data', sa.String(30)), PrimaryKeyConstraint( "id", name="id_pk", comment="id_pk comment" ), CheckConstraint('id < 100', name="cc1", comment="Id value can't exceed 100"), UniqueConstraint(['data'], name="uc1", comment="Must have unique data field"), ) ``` Provides the DDL representation for constraint comments and routines to create and drop them. Class `.Inspector` reflects constraint comments via methods like `get_check_constraints` . ### Checklist <!-- go over following points. check them with an `x` if they do apply, (they turn into clickable checkboxes once the PR is submitted, so no need to do everything at once) --> This pull request is: - [ ] A documentation / typographical error fix - [ ] A short code fix - [x] A new feature implementation - Solves the issue 5667. - The commit message includes `Fixes: 5667`. - Includes tests based on comment reflection. **Have a nice day!** Fixes: #5667 Closes: #7742 Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/7742 Pull-request-sha: 42a5d3c3e9ccf9a9d5397fd007aeab0854f66130 Change-Id: Ia60f578595afdbd6089541c9a00e37997ef78ad3
Diffstat (limited to 'test/dialect/postgresql/test_reflection.py')
-rw-r--r--test/dialect/postgresql/test_reflection.py104
1 files changed, 96 insertions, 8 deletions
diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index 99bc14d78..6bcd7a87c 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -32,6 +32,7 @@ from sqlalchemy.dialects.postgresql import TSRANGE
from sqlalchemy.engine import ObjectKind
from sqlalchemy.engine import ObjectScope
from sqlalchemy.schema import CreateIndex
+from sqlalchemy.sql import ddl as sa_ddl
from sqlalchemy.sql.schema import CheckConstraint
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import fixtures
@@ -1496,6 +1497,7 @@ class ReflectionTest(
"initially": "DEFERRED",
"match": "FULL",
},
+ "comment": None,
},
"company_industry_id_fkey": {
"name": "company_industry_id_fkey",
@@ -1504,6 +1506,7 @@ class ReflectionTest(
"referred_table": "industry",
"referred_schema": None,
"options": {"onupdate": "CASCADE", "ondelete": "CASCADE"},
+ "comment": None,
},
}
metadata.create_all(connection)
@@ -1928,7 +1931,7 @@ class ReflectionTest(
)
def test_reflect_check_warning(self):
- rows = [("foo", "some name", "NOTCHECK foobar")]
+ rows = [("foo", "some name", "NOTCHECK foobar", None)]
conn = mock.Mock(
execute=lambda *arg, **kw: mock.MagicMock(
fetchall=lambda: rows, __iter__=lambda self: iter(rows)
@@ -1941,10 +1944,20 @@ class ReflectionTest(
def test_reflect_extra_newlines(self):
rows = [
- ("foo", "some name", "CHECK (\n(a \nIS\n NOT\n\n NULL\n)\n)"),
- ("foo", "some other name", "CHECK ((b\nIS\nNOT\nNULL))"),
- ("foo", "some CRLF name", "CHECK ((c\r\n\r\nIS\r\nNOT\r\nNULL))"),
- ("foo", "some name", "CHECK (c != 'hi\nim a name\n')"),
+ (
+ "foo",
+ "some name",
+ "CHECK (\n(a \nIS\n NOT\n\n NULL\n)\n)",
+ None,
+ ),
+ ("foo", "some other name", "CHECK ((b\nIS\nNOT\nNULL))", None),
+ (
+ "foo",
+ "some CRLF name",
+ "CHECK ((c\r\n\r\nIS\r\nNOT\r\nNULL))",
+ None,
+ ),
+ ("foo", "some name", "CHECK (c != 'hi\nim a name\n')", None),
]
conn = mock.Mock(
execute=lambda *arg, **kw: mock.MagicMock(
@@ -1960,18 +1973,30 @@ class ReflectionTest(
{
"name": "some name",
"sqltext": "a \nIS\n NOT\n\n NULL\n",
+ "comment": None,
+ },
+ {
+ "name": "some other name",
+ "sqltext": "b\nIS\nNOT\nNULL",
+ "comment": None,
},
- {"name": "some other name", "sqltext": "b\nIS\nNOT\nNULL"},
{
"name": "some CRLF name",
"sqltext": "c\r\n\r\nIS\r\nNOT\r\nNULL",
+ "comment": None,
+ },
+ {
+ "name": "some name",
+ "sqltext": "c != 'hi\nim a name\n'",
+ "comment": None,
},
- {"name": "some name", "sqltext": "c != 'hi\nim a name\n'"},
],
)
def test_reflect_with_not_valid_check_constraint(self):
- rows = [("foo", "some name", "CHECK ((a IS NOT NULL)) NOT VALID")]
+ rows = [
+ ("foo", "some name", "CHECK ((a IS NOT NULL)) NOT VALID", None)
+ ]
conn = mock.Mock(
execute=lambda *arg, **kw: mock.MagicMock(
fetchall=lambda: rows, __iter__=lambda self: iter(rows)
@@ -1987,6 +2012,7 @@ class ReflectionTest(
"name": "some name",
"sqltext": "a IS NOT NULL",
"dialect_options": {"not_valid": True},
+ "comment": None,
}
],
)
@@ -2069,6 +2095,68 @@ class ReflectionTest(
[["b"]],
)
+ def test_reflection_constraint_comments(self, connection, metadata):
+ t = Table(
+ "foo",
+ metadata,
+ Column("id", Integer),
+ Column("foo_id", ForeignKey("foo.id", name="fk_1")),
+ Column("foo_other_id", ForeignKey("foo.id", name="fk_2")),
+ CheckConstraint("id>0", name="ch_1"),
+ CheckConstraint("id<1000", name="ch_2"),
+ PrimaryKeyConstraint("id", name="foo_pk"),
+ UniqueConstraint("id", "foo_id", name="un_1"),
+ UniqueConstraint("id", "foo_other_id", name="un_2"),
+ )
+ metadata.create_all(connection)
+
+ def check(elements, exp):
+ elements = {c["name"]: c["comment"] for c in elements}
+ eq_(elements, exp)
+
+ def all_none():
+ insp = inspect(connection)
+ is_(insp.get_pk_constraint("foo")["comment"], None)
+ check(
+ insp.get_check_constraints("foo"), {"ch_1": None, "ch_2": None}
+ )
+ check(
+ insp.get_unique_constraints("foo"),
+ {"un_1": None, "un_2": None},
+ )
+ check(insp.get_foreign_keys("foo"), {"fk_1": None, "fk_2": None})
+
+ all_none()
+
+ c = next(c for c in t.constraints if c.name == "ch_1")
+ u = next(c for c in t.constraints if c.name == "un_1")
+ f = next(c for c in t.foreign_key_constraints if c.name == "fk_1")
+ p = t.primary_key
+ c.comment = "cc comment"
+ u.comment = "uc comment"
+ f.comment = "fc comment"
+ p.comment = "pk comment"
+ for cst in [c, u, f, p]:
+ connection.execute(sa_ddl.SetConstraintComment(cst))
+
+ insp = inspect(connection)
+ eq_(insp.get_pk_constraint("foo")["comment"], "pk comment")
+ check(
+ insp.get_check_constraints("foo"),
+ {"ch_1": "cc comment", "ch_2": None},
+ )
+ check(
+ insp.get_unique_constraints("foo"),
+ {"un_1": "uc comment", "un_2": None},
+ )
+ check(
+ insp.get_foreign_keys("foo"), {"fk_1": "fc comment", "fk_2": None}
+ )
+
+ for cst in [c, u, f, p]:
+ connection.execute(sa_ddl.DropConstraintComment(cst))
+ all_none()
+
class CustomTypeReflectionTest(fixtures.TestBase):
class CustomType: