"""
def configure(
- connection: Optional["Connection"],
- url: Optional[str],
- dialect_name: Optional[str],
- dialect_opts: Optional[dict],
- transactional_ddl: Optional[bool],
- transaction_per_migration: bool,
- output_buffer: Optional[TextIO],
- starting_rev: Optional[str],
- tag: Optional[str],
- template_args: Optional[dict],
- render_as_batch: bool,
- target_metadata: Optional["MetaData"],
- include_name: Optional[Callable],
- include_object: Optional[Callable],
- include_schemas: bool,
- process_revision_directives: Optional[Callable],
- compare_type: bool,
- compare_server_default: bool,
- render_item: Optional[Callable],
- literal_binds: bool,
- upgrade_token: str,
- downgrade_token: str,
- alembic_module_prefix: str,
- sqlalchemy_module_prefix: str,
- user_module_prefix: Optional[str],
- on_version_apply: Optional[Callable],
+ connection: Optional["Connection"] = None,
+ url: Optional[str] = None,
+ dialect_name: Optional[str] = None,
+ dialect_opts: Optional[dict] = None,
+ transactional_ddl: Optional[bool] = None,
+ transaction_per_migration: bool = False,
+ output_buffer: Optional[TextIO] = None,
+ starting_rev: Optional[str] = None,
+ tag: Optional[str] = None,
+ template_args: Optional[dict] = None,
+ render_as_batch: bool = False,
+ target_metadata: Optional["MetaData"] = None,
+ include_name: Optional[Callable] = None,
+ include_object: Optional[Callable] = None,
+ include_schemas: bool = False,
+ process_revision_directives: Optional[Callable] = None,
+ compare_type: bool = False,
+ compare_server_default: bool = False,
+ render_item: Optional[Callable] = None,
+ literal_binds: bool = False,
+ upgrade_token: str = "upgrades",
+ downgrade_token: str = "downgrades",
+ alembic_module_prefix: str = "op.",
+ sqlalchemy_module_prefix: str = "sa.",
+ user_module_prefix: Optional[str] = None,
+ on_version_apply: Optional[Callable] = None,
**kw
) -> None:
"""Configure a :class:`.MigrationContext` within this
"""
-def execute(sql, execution_options):
+def execute(sql, execution_options=None):
"""Execute the given SQL using the current change context.
The behavior of :meth:`.execute` is the same
"""
-def get_x_argument(as_dictionary: bool):
+def get_x_argument(as_dictionary: bool = False):
"""Return the value(s) passed for the ``-x`` argument, if any.
The ``-x`` argument is an open ended flag that allows any user-defined
from sqlalchemy.sql.schema import Identity
from sqlalchemy.sql.schema import Table
from sqlalchemy.sql.type_api import TypeEngine
+ from sqlalchemy.util import immutabledict
from .operations.ops import MigrateOperation
from .util.sqla_compat import _literal_bindparam
### end imports ###
def add_column(
- table_name: str, column: "Column", schema: Optional[str]
+ table_name: str, column: "Column", schema: Optional[str] = None
) -> Optional["Table"]:
"""Issue an "add column" instruction using the current
migration context.
def alter_column(
table_name: str,
column_name: str,
- nullable: Optional[bool],
- comment: Union[str, bool, None],
- server_default: Any,
- new_column_name: Optional[str],
- type_: Union["TypeEngine", Type["TypeEngine"], None],
- existing_type: Union["TypeEngine", Type["TypeEngine"], None],
- existing_server_default: Union[str, bool, "Identity", "Computed", None],
- existing_nullable: Optional[bool],
- existing_comment: Optional[str],
- schema: Optional[str],
+ nullable: Optional[bool] = None,
+ comment: Union[str, bool, None] = False,
+ server_default: Any = False,
+ new_column_name: Optional[str] = None,
+ type_: Union["TypeEngine", Type["TypeEngine"], None] = None,
+ existing_type: Union["TypeEngine", Type["TypeEngine"], None] = None,
+ existing_server_default: Union[
+ str, bool, "Identity", "Computed", None
+ ] = False,
+ existing_nullable: Optional[bool] = None,
+ existing_comment: Optional[str] = None,
+ schema: Optional[str] = None,
**kw
) -> Optional["Table"]:
"""Issue an "alter column" instruction using the
def batch_alter_table(
table_name,
- schema,
- recreate,
- partial_reordering,
- copy_from,
- table_args,
- table_kwargs,
- reflect_args,
- reflect_kwargs,
- naming_convention,
+ schema=None,
+ recreate="auto",
+ partial_reordering=None,
+ copy_from=None,
+ table_args=(),
+ table_kwargs=immutabledict({}),
+ reflect_args=(),
+ reflect_kwargs=immutabledict({}),
+ naming_convention=None,
):
"""Invoke a series of per-table migrations in batch.
"""
def bulk_insert(
- table: Union["Table", "TableClause"], rows: List[dict], multiinsert: bool
+ table: Union["Table", "TableClause"],
+ rows: List[dict],
+ multiinsert: bool = True,
) -> None:
"""Issue a "bulk insert" operation using the current
migration context.
constraint_name: Optional[str],
table_name: str,
condition: "BinaryExpression",
- schema: Optional[str],
+ schema: Optional[str] = None,
**kw
) -> Optional["Table"]:
"""Issue a "create check constraint" instruction using the
referent_table: str,
local_cols: List[str],
remote_cols: List[str],
- onupdate: Optional[str],
- ondelete: Optional[str],
- deferrable: Optional[bool],
- initially: Optional[str],
- match: Optional[str],
- source_schema: Optional[str],
- referent_schema: Optional[str],
+ onupdate: Optional[str] = None,
+ ondelete: Optional[str] = None,
+ deferrable: Optional[bool] = None,
+ initially: Optional[str] = None,
+ match: Optional[str] = None,
+ source_schema: Optional[str] = None,
+ referent_schema: Optional[str] = None,
**dialect_kw
) -> Optional["Table"]:
"""Issue a "create foreign key" instruction using the
index_name: str,
table_name: str,
columns: Sequence[Union[str, "TextClause", "Function"]],
- schema: Optional[str],
- unique: bool,
+ schema: Optional[str] = None,
+ unique: bool = False,
**kw
) -> Optional["Table"]:
"""Issue a "create index" instruction using the current
constraint_name: str,
table_name: str,
columns: List[str],
- schema: Optional[str],
+ schema: Optional[str] = None,
) -> Optional["Table"]:
"""Issue a "create primary key" instruction using the current
migration context.
def create_table_comment(
table_name: str,
comment: Optional[str],
- existing_comment: None,
- schema: Optional[str],
+ existing_comment: None = None,
+ schema: Optional[str] = None,
) -> Optional["Table"]:
"""Emit a COMMENT ON operation to set the comment for a table.
constraint_name: Optional[str],
table_name: str,
columns: Sequence[str],
- schema: Optional[str],
+ schema: Optional[str] = None,
**kw
) -> Any:
"""Issue a "create unique constraint" instruction using the
"""
def drop_column(
- table_name: str, column_name: str, schema: Optional[str], **kw
+ table_name: str, column_name: str, schema: Optional[str] = None, **kw
) -> Optional["Table"]:
"""Issue a "drop column" instruction using the current
migration context.
def drop_constraint(
constraint_name: str,
table_name: str,
- type_: Optional[str],
- schema: Optional[str],
+ type_: Optional[str] = None,
+ schema: Optional[str] = None,
) -> Optional["Table"]:
"""Drop a constraint of the given name, typically via DROP CONSTRAINT.
"""
def drop_index(
- index_name: str, table_name: Optional[str], schema: Optional[str], **kw
+ index_name: str,
+ table_name: Optional[str] = None,
+ schema: Optional[str] = None,
+ **kw
) -> Optional["Table"]:
"""Issue a "drop index" instruction using the current
migration context.
"""
-def drop_table(table_name: str, schema: Optional[str], **kw: Any) -> None:
+def drop_table(
+ table_name: str, schema: Optional[str] = None, **kw: Any
+) -> None:
"""Issue a "drop table" instruction using the current
migration context.
"""
def drop_table_comment(
- table_name: str, existing_comment: Optional[str], schema: Optional[str]
+ table_name: str,
+ existing_comment: Optional[str] = None,
+ schema: Optional[str] = None,
) -> Optional["Table"]:
"""Issue a "drop table comment" operation to
remove an existing comment set on a table.
"""
def execute(
- sqltext: Union[str, "TextClause", "Update"], execution_options: None
+ sqltext: Union[str, "TextClause", "Update"], execution_options: None = None
) -> Optional["Table"]:
"""Execute the given SQL using the current migration context.
"""
def inline_literal(
- value: Union[str, int], type_: None
+ value: Union[str, int], type_: None = None
) -> "_literal_bindparam":
"""Produce an 'inline literal' expression, suitable for
using in an INSERT, UPDATE, or DELETE statement.
"""
-def register_operation(name: str, sourcename: Optional[str]) -> Callable:
+def register_operation(
+ name: str, sourcename: Optional[str] = None
+) -> Callable:
"""Register a new operation for this class.
This method is normally used to add new operations
"""
def rename_table(
- old_table_name: str, new_table_name: str, schema: Optional[str]
+ old_table_name: str, new_table_name: str, schema: Optional[str] = None
) -> Optional["Table"]:
"""Emit an ALTER TABLE to rename a table.
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
