--- /dev/null
+.. change::
+ :tags: usecase, postgresql
+ :tickets: 11122, 13432
+
+ Added ``postgresql_with`` support to :class:`.CreateView` for specifying
+ PostgreSQL view options such as ``security_invoker``, ``security_barrier``,
+ and ``check_option``, rendered as a ``WITH (...)`` clause between the view
+ name and the ``AS`` keyword. Additionally, the ``postgresql_with`` parameter
+ accepted by :class:`_schema.Table` and :class:`_schema.Index` now correctly
+ renders Python boolean values as ``true``/``false`` (lowercase), and
+ ``None`` values as the parameter name alone without an ``= value`` portion.
+ Pull request courtesy alphavector.
Column("id", Integer, primary_key=True),
postgresql_with={
"fillfactor": 70,
- "autovacuum_enabled": "false",
+ "autovacuum_enabled": False,
"toast.vacuum_truncate": True,
},
)
CREATE TABLE mytable (
id INTEGER NOT NULL,
PRIMARY KEY (id)
- ) WITH (fillfactor = 70, autovacuum_enabled = false, toast.vacuum_truncate = True)
+ ) WITH (fillfactor = 70, autovacuum_enabled = false, toast.vacuum_truncate = true)
-The values in the dictionary can be integers, strings, or boolean values.
+The values in the dictionary can be integers, strings, boolean values, or
+``None``. Boolean values are rendered as lowercase ``true``/``false``.
+A ``None`` value renders the parameter name only, without an ``=`` value.
Parameter names can include dots to specify parameters for associated objects
like toast tables (e.g., ``toast.vacuum_truncate``).
Table("some_table", metadata, ..., postgresql_with_oids=False)
+.. _postgresql_view_options:
+
+PostgreSQL View Options
+-----------------------
+
+The following sections indicate options which are supported by the PostgreSQL
+dialect in conjunction with :class:`.CreateView`.
+
+``WITH``
+^^^^^^^^
+
+Specifies view-level options for :class:`.CreateView` using the ``WITH``
+clause::
+
+ from sqlalchemy import CreateView, select
+
+ CreateView(
+ select(my_table),
+ "my_view",
+ postgresql_with={"security_invoker": True},
+ )
+
+This generates:
+
+.. sourcecode:: sql
+
+ CREATE VIEW my_view WITH (security_invoker = true) AS SELECT ...
+
+The same formatting rules that apply to :ref:`postgresql_table_options_with`
+apply here: booleans render as lowercase ``true``/``false``, ``None``
+renders as the parameter name only, and other values render as-is.
+
+.. versionadded:: 2.1
+
.. _postgresql_constraint_options:
PostgreSQL Constraint Options
domain = drop.element
return f"DROP DOMAIN {self.preparer.format_type(domain)}"
+ def _prepare_withclause_opts(self, withclause):
+ with_opts = []
+ for param, value in withclause.items():
+ if value is None:
+ with_opts.append(param)
+ elif isinstance(value, bool):
+ with_opts.append(
+ "%s = %s" % (param, "true" if value else "false")
+ )
+ else:
+ with_opts.append("%s = %s" % (param, value))
+ return ", ".join(with_opts)
+
+ def create_table_select_suffixes(self, element, type_, **kw):
+ if type_ == "view":
+ withclause = element.dialect_options["postgresql"]["with"]
+ if withclause:
+ return "WITH (%s)" % self._prepare_withclause_opts(withclause)
+ return ""
+
def visit_create_index(self, create, **kw):
preparer = self.preparer
index = create.element
withclause = index.dialect_options["postgresql"]["with"]
if withclause:
- text += " WITH (%s)" % (
- ", ".join(
- [
- "%s = %s" % storage_parameter
- for storage_parameter in withclause.items()
- ]
- )
- )
+ text += " WITH (%s)" % self._prepare_withclause_opts(withclause)
tablespace_name = index.dialect_options["postgresql"]["tablespace"]
if tablespace_name:
table_opts.append("\n USING %s" % pg_opts["using"])
if pg_opts["with"]:
- storage_params = (f"{k} = {v}" for k, v in pg_opts["with"].items())
- table_opts.append(f" WITH ({', '.join(storage_params)})")
+ table_opts.append(
+ "\n WITH (%s)" % self._prepare_withclause_opts(pg_opts["with"])
+ )
if pg_opts["with_oids"] is True:
table_opts.append("\n WITH OIDS")
"nulls_not_distinct": None,
},
),
+ (
+ schema.CreateView,
+ {
+ "with": None,
+ },
+ ),
]
reflection_options = ("postgresql_ignore_search_path",)
def visit_create_table_as(self, element: CreateTableAs, **kw: Any) -> str:
return self._generate_table_select(element, "create_table_as", **kw)
+ def create_table_select_suffixes(
+ self,
+ element: _TableViaSelect,
+ type_: str,
+ **kw: Any,
+ ) -> str:
+ return ""
+
def _generate_table_select(
self,
element: _TableViaSelect,
else element.if_not_exists
)
- parts = [
+ parts: List[Optional[str]] = [
"CREATE",
"OR REPLACE" if getattr(element, "or_replace", False) else None,
"TEMPORARY" if element.temporary else None,
),
"IF NOT EXISTS" if use_if_not_exists else None,
prep.format_table(element.table),
- "AS",
- select_sql,
]
+ suffixes = self.create_table_select_suffixes(element, type_, **kw)
+ if suffixes:
+ parts.append(suffixes)
+ parts += ["AS", select_sql]
return " ".join(p for p in parts if p)
def visit_create_column(self, create, first_pk=False, **kw):
from sqlalchemy import CheckConstraint
from sqlalchemy import Column
from sqlalchemy import Computed
+from sqlalchemy import CreateView
from sqlalchemy import Date
from sqlalchemy import delete
from sqlalchemy import Enum
self.assert_compile(
schema.CreateTable(tbl3),
"CREATE TABLE atable3 () "
- "WITH (user_catalog_table = False, parallel_workers = 15)",
+ "WITH (user_catalog_table = false, parallel_workers = 15)",
+ )
+
+ tbl4 = Table(
+ "atable4",
+ m,
+ Column("id", Integer),
+ postgresql_with={
+ "autovacuum_enabled": True,
+ "autovacuum_analyze_scale_factor": 0.2,
+ "vacuum_index_cleanup": "auto",
+ "vacuum_truncate": None,
+ },
+ )
+
+ self.assert_compile(
+ schema.CreateTable(tbl4),
+ "CREATE TABLE atable4 (id INTEGER) "
+ "WITH (autovacuum_enabled = true, "
+ "autovacuum_analyze_scale_factor = 0.2, "
+ "vacuum_index_cleanup = auto, "
+ "vacuum_truncate)",
)
def test_create_table_with_oncommit_option(self):
"ON COMMIT PRESERVE ROWS TABLESPACE sometablespace",
)
+ def test_create_view_with_options(self):
+ src = table("src", column("id"), column("name"))
+
+ stmt = CreateView(
+ select(src.c.id, src.c.name),
+ "my_view",
+ postgresql_with={"security_invoker": True},
+ )
+ self.assert_compile(
+ stmt,
+ "CREATE VIEW my_view WITH (security_invoker = true) "
+ "AS SELECT src.id, src.name FROM src",
+ )
+
+ stmt2 = CreateView(
+ select(src.c.id),
+ "other_view",
+ postgresql_with={
+ "security_barrier": True,
+ "check_option": "local",
+ },
+ )
+ self.assert_compile(
+ stmt2,
+ "CREATE VIEW other_view "
+ "WITH (security_barrier = true, check_option = local) "
+ "AS SELECT src.id FROM src",
+ )
+
def test_create_partial_index(self):
m = MetaData()
tbl = Table("testtbl", m, Column("data", Integer))
postgresql_with={"buffering": "off"},
)
+ idx4 = Index(
+ "test_idx4",
+ tbl.c.data,
+ postgresql_using="gin",
+ postgresql_with={
+ "fastupdate": False,
+ "gin_pending_list_limit": 4096,
+ },
+ )
+
+ idx5 = Index(
+ "test_idx5",
+ tbl.c.data,
+ postgresql_using="brin",
+ postgresql_with={
+ "pages_per_range": 1,
+ "autosummarize": None,
+ },
+ )
+
self.assert_compile(
schema.CreateIndex(idx1),
"CREATE INDEX test_idx1 ON testtbl (data)",
"USING gist (data) "
"WITH (buffering = off)",
)
+ self.assert_compile(
+ schema.CreateIndex(idx4),
+ "CREATE INDEX test_idx4 ON testtbl "
+ "USING gin (data) "
+ "WITH (fastupdate = false, gin_pending_list_limit = 4096)",
+ )
+ self.assert_compile(
+ schema.CreateIndex(idx5),
+ "CREATE INDEX test_idx5 ON testtbl "
+ "USING brin (data) "
+ "WITH (pages_per_range = 1, autosummarize)",
+ )
def test_create_index_with_using_unusual_conditions(self):
m = MetaData()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
