return self.sql_compiler.post_process_text(ddl.statement % context)
def visit_create_schema(self, create, **kw):
- schema = self.preparer.format_schema(create.element)
- return "CREATE SCHEMA " + schema
+ text = "CREATE SCHEMA "
+ if create.if_not_exists:
+ text += "IF NOT EXISTS "
+ return text + self.preparer.format_schema(create.element)
def visit_drop_schema(self, drop, **kw):
- schema = self.preparer.format_schema(drop.element)
- text = "DROP SCHEMA " + schema
+ text = "DROP SCHEMA "
+ if drop.if_exists:
+ text += "IF EXISTS "
+ text += self.preparer.format_schema(drop.element)
if drop.cascade:
text += " CASCADE"
return text
return " ".join(text)
def visit_create_sequence(self, create, prefix=None, **kw):
- text = "CREATE SEQUENCE %s" % self.preparer.format_sequence(
- create.element
- )
+ text = "CREATE SEQUENCE "
+ if create.if_not_exists:
+ text += "IF NOT EXISTS "
+ text += self.preparer.format_sequence(create.element)
+
if prefix:
text += prefix
if create.element.start is None:
return text
def visit_drop_sequence(self, drop, **kw):
- return "DROP SEQUENCE %s" % self.preparer.format_sequence(drop.element)
+ text = "DROP SEQUENCE "
+ if drop.if_exists:
+ text += "IF EXISTS "
+ return text + self.preparer.format_sequence(drop.element)
def visit_drop_constraint(self, drop, **kw):
constraint = drop.element
from .schema import Constraint
from .schema import ForeignKeyConstraint
from .schema import SchemaItem
+ from .schema import Sequence
from .schema import Table
from ..engine.base import _CompiledCacheType
from ..engine.base import Connection
def __init__(
self,
element,
- if_exists=False,
- if_not_exists=False,
):
self.element = self.target = element
- self.if_exists = if_exists
- self.if_not_exists = if_not_exists
self._ddl_if = getattr(element, "_ddl_if", None)
@property
return False
-class CreateSchema(_CreateDropBase):
+class _CreateBase(_CreateDropBase):
+ def __init__(self, element, if_not_exists=False):
+ super().__init__(element)
+ self.if_not_exists = if_not_exists
+
+
+class _DropBase(_CreateDropBase):
+ def __init__(self, element, if_exists=False):
+ super().__init__(element)
+ self.if_exists = if_exists
+
+
+class CreateSchema(_CreateBase):
"""Represent a CREATE SCHEMA statement.
The argument here is the string name of the schema.
def __init__(
self,
name,
- quote=None,
- if_exists=False,
if_not_exists=False,
):
"""Create a new :class:`.CreateSchema` construct."""
- self.quote = quote
- self.element = name
- self.if_exists = if_exists
- self.if_not_exists = if_not_exists
+ super().__init__(element=name, if_not_exists=if_not_exists)
-class DropSchema(_CreateDropBase):
+class DropSchema(_DropBase):
"""Represent a DROP SCHEMA statement.
The argument here is the string name of the schema.
def __init__(
self,
name,
- quote=None,
cascade=False,
if_exists=False,
- if_not_exists=False,
):
"""Create a new :class:`.DropSchema` construct."""
- self.quote = quote
+ super().__init__(element=name, if_exists=if_exists)
self.cascade = cascade
- self.quote = quote
- self.element = name
- self.if_exists = if_exists
- self.if_not_exists = if_not_exists
-class CreateTable(_CreateDropBase):
+class CreateTable(_CreateBase):
"""Represent a CREATE TABLE statement."""
__visit_name__ = "create_table"
self.include_foreign_key_constraints = include_foreign_key_constraints
-class _DropView(_CreateDropBase):
+class _DropView(_DropBase):
"""Semi-public 'DROP VIEW' construct.
Used by the test suite for dialect-agnostic drops of views.
self.element = element
-class DropTable(_CreateDropBase):
+class DropTable(_DropBase):
"""Represent a DROP TABLE statement."""
__visit_name__ = "drop_table"
super().__init__(element, if_exists=if_exists)
-class CreateSequence(_CreateDropBase):
+class CreateSequence(_CreateBase):
"""Represent a CREATE SEQUENCE statement."""
__visit_name__ = "create_sequence"
+ def __init__(self, element: Sequence, if_not_exists: bool = False):
+ super().__init__(element, if_not_exists=if_not_exists)
+
-class DropSequence(_CreateDropBase):
+class DropSequence(_DropBase):
"""Represent a DROP SEQUENCE statement."""
__visit_name__ = "drop_sequence"
+ def __init__(self, element: Sequence, if_exists: bool = False):
+ super().__init__(element, if_exists=if_exists)
+
-class CreateIndex(_CreateDropBase):
+class CreateIndex(_CreateBase):
"""Represent a CREATE INDEX statement."""
__visit_name__ = "create_index"
:param element: a :class:`_schema.Index` that's the subject
of the CREATE.
- :param on: See the description for 'on' in :class:`.DDL`.
:param if_not_exists: if True, an IF NOT EXISTS operator will be
applied to the construct.
super().__init__(element, if_not_exists=if_not_exists)
-class DropIndex(_CreateDropBase):
+class DropIndex(_DropBase):
"""Represent a DROP INDEX statement."""
__visit_name__ = "drop_index"
:param element: a :class:`_schema.Index` that's the subject
of the DROP.
- :param on: See the description for 'on' in :class:`.DDL`.
:param if_exists: if True, an IF EXISTS operator will be applied to the
construct.
super().__init__(element, if_exists=if_exists)
-class AddConstraint(_CreateDropBase):
+class AddConstraint(_CreateBase):
"""Represent an ALTER TABLE ADD CONSTRAINT statement."""
__visit_name__ = "add_constraint"
- def __init__(self, element, *args, **kw):
- super().__init__(element, *args, **kw)
+ def __init__(self, element):
+ super().__init__(element)
element._create_rule = util.portable_instancemethod(
self._create_rule_disable
)
-class DropConstraint(_CreateDropBase):
+class DropConstraint(_DropBase):
"""Represent an ALTER TABLE DROP CONSTRAINT statement."""
__visit_name__ = "drop_constraint"
- def __init__(self, element, cascade=False, **kw):
+ def __init__(self, element, cascade=False, if_exists=False, **kw):
self.cascade = cascade
- super().__init__(element, **kw)
+ super().__init__(element, if_exists=if_exists, **kw)
element._create_rule = util.portable_instancemethod(
self._create_rule_disable
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
