from .. import util
from ..operations import ops
-from ..util import compat
from ..util import sqla_compat
-from ..util.compat import string_types
if TYPE_CHECKING:
from typing import Literal
if name is None:
return name
elif isinstance(name, sql.elements.quoted_name):
- return compat.text_type(name)
- elif isinstance(name, compat.string_types):
+ return str(name)
+ elif isinstance(name, str):
return name
elif sqla_compat._server_default_is_identity(default):
return _render_identity(cast("Identity", default), autogen_context)
elif isinstance(default, sa_schema.DefaultClause):
- if isinstance(default.arg, compat.string_types):
+ if isinstance(default.arg, str):
default = default.arg
else:
return _render_potential_expr(
default.arg, autogen_context, is_server_default=True
)
- if isinstance(default, string_types) and repr_:
+ if isinstance(default, str) and repr_:
default = repr(re.sub(r"^'|'$", "", default))
return cast(str, default)
def _execute_sql(
autogen_context: "AutogenContext", op: "ops.ExecuteSQLOp"
) -> str:
- if not isinstance(op.sqltext, string_types):
+ if not isinstance(op.sqltext, str):
raise NotImplementedError(
"Autogenerate rendering of SQL Expression language constructs "
"not supported here; please use a plain SQL string"
"""
if arg:
- output = compat.text_type(text) % arg
+ output = str(text) % arg
else:
- output = compat.text_type(text)
+ output = str(text)
util.write_outstream(self.stdout, output, "\n")
from . import base
from .. import util
from ..util import sqla_compat
-from ..util.compat import string_types
-from ..util.compat import text_type
if TYPE_CHECKING:
from io import StringIO
def static_output(self, text: str) -> None:
assert self.output_buffer is not None
- self.output_buffer.write(text_type(text + "\n\n"))
+ self.output_buffer.write(text + "\n\n")
self.output_buffer.flush()
def requires_recreate_in_batch(
multiparams: Sequence[dict] = (),
params: Dict[str, int] = util.immutabledict(),
) -> Optional[Union["LegacyCursorResult", "CursorResult"]]:
- if isinstance(construct, string_types):
+ if isinstance(construct, str):
construct = text(construct)
if self.as_sql:
if multiparams or params:
compile_kw = {}
self.static_output(
- text_type(
- construct.compile(dialect=self.dialect, **compile_kw)
- )
+ str(construct.compile(dialect=self.dialect, **compile_kw))
.replace("\t", " ")
.strip()
+ self.command_terminator
def correct_for_autogen_constraints(
self,
- conn_uniques: Union[Set["UniqueConstraint"]],
- conn_indexes: Union[Set["Index"]],
+ conn_uniques: Set["UniqueConstraint"],
+ conn_indexes: Set["Index"],
metadata_unique_constraints: Set["UniqueConstraint"],
metadata_indexes: Set["Index"],
) -> None:
compile_kw = dict(
compile_kwargs={"literal_binds": True, "include_table": False}
)
- return text_type(expr.compile(dialect=self.dialect, **compile_kw))
+ return str(expr.compile(dialect=self.dialect, **compile_kw))
def _compat_autogen_column_reflect(
self, inspector: "Inspector"
from ..operations import schemaobj
from ..operations.base import BatchOperations
from ..operations.base import Operations
-from ..util import compat
from ..util import sqla_compat
if TYPE_CHECKING:
if (
not isinstance(inspector_column.type, Numeric)
and metadata_column.server_default is not None
- and isinstance(
- metadata_column.server_default.arg, compat.string_types
- )
+ and isinstance(metadata_column.server_default.arg, str)
and not re.match(r"^'.*'$", rendered_metadata_default)
):
rendered_metadata_default = "'%s'" % rendered_metadata_default
from .. import util
from ..util import sqla_compat
-from ..util.compat import string_types
if TYPE_CHECKING:
from sqlalchemy.sql.elements import ColumnElement
ForeignKey.
"""
- if isinstance(fk._colspec, string_types): # type:ignore[attr-defined]
+ if isinstance(fk._colspec, str): # type:ignore[attr-defined]
table_key, cname = fk._colspec.rsplit( # type:ignore[attr-defined]
".", 1
)
from sqlalchemy import util as sqlautil
from .. import util
-from ..util import compat
if TYPE_CHECKING:
from typing import Literal
self, id_: Optional[str]
) -> Tuple[Tuple[str, ...], Optional[str]]:
branch_label: Optional[str]
- if isinstance(id_, compat.string_types) and "@" in id_:
+ if isinstance(id_, str) and "@" in id_:
branch_label, id_ = id_.split("@", 1)
elif id_ is not None and (
- (
- isinstance(id_, tuple)
- and id_
- and not isinstance(id_[0], compat.string_types)
- )
- or not isinstance(id_, compat.string_types + (tuple,))
+ (isinstance(id_, tuple) and id_ and not isinstance(id_[0], str))
+ or not isinstance(id_, (str, tuple))
):
raise RevisionError(
"revision identifier %r is not a string; ensure database "
walk to.
"""
initial: Optional[_RevisionOrBase]
- if isinstance(start, compat.string_types):
+ if isinstance(start, str):
initial = self.get_revision(start)
else:
initial = start
if target is None:
return None, None
assert isinstance(
- target, compat.string_types
+ target, str
), "Expected downgrade target in string form"
match = _relative_destination.match(target)
if match:
to. The target may be specified in absolute form, or relative to
:current_revisions.
"""
- if isinstance(target, compat.string_types):
+ if isinstance(target, str):
match = _relative_destination.match(target)
else:
match = None
# Handled named bases (e.g. branch@... -> heads should only produce
# targets on the given branch)
- if isinstance(lower, compat.string_types) and "@" in lower:
+ if isinstance(lower, str) and "@" in lower:
branch, _, _ = lower.partition("@")
branch_rev = self.get_revision(branch)
if branch_rev is not None and branch_rev.revision == branch:
from typing import Dict
from sqlalchemy import exc as sa_exc
-from sqlalchemy import util
from sqlalchemy.engine import default
from sqlalchemy.testing.assertions import _expect_warnings
from sqlalchemy.testing.assertions import eq_ # noqa
ec.error = err
success = True
if msg is not None:
- assert re.search(
- msg, util.text_type(err), re.UNICODE
- ), "%r !~ %s" % (msg, err)
+ assert re.search(msg, str(err), re.UNICODE), f"{msg} !~ {err}"
if check_context and not are_we_already_in_a_traceback:
_assert_proper_exception_context(err)
- print(util.text_type(err).encode("utf-8"))
+ print(str(err).encode("utf-8"))
# assert outside the block so it works for AssertionError too !
assert success, "Callable did not raise an exception"
from ..migration import MigrationContext
from ..operations import Operations
from ..util import sqla_compat
-from ..util.compat import string_types
-from ..util.compat import text_type
from ..util.sqla_compat import create_mock_engine
from ..util.sqla_compat import sqla_14
from ..util.sqla_compat import sqla_1x
if not as_sql:
def execute(stmt, *multiparam, **param):
- if isinstance(stmt, string_types):
+ if isinstance(stmt, str):
stmt = text(stmt)
assert stmt.supports_execution
- sql = text_type(stmt.compile(dialect=ctx_dialect))
+ sql = str(stmt.compile(dialect=ctx_dialect))
buf.write(sql)
+from itertools import zip_longest
+
from sqlalchemy import schema
-from sqlalchemy import util
class CompareTable:
if self.table.name != other.name or self.table.schema != other.schema:
return False
- for c1, c2 in util.zip_longest(self.table.c, other.c):
+ for c1, c2 in zip_longest(self.table.c, other.c):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
)
if not r1:
return False
- for c1, c2 in util.zip_longest(self.constraint.columns, other.columns):
+ for c1, c2 in zip_longest(self.constraint.columns, other.columns):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
if not r1:
return False
- for c1, c2 in util.zip_longest(self.constraint.columns, other.columns):
+ for c1, c2 in zip_longest(self.constraint.columns, other.columns):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
if not r1:
return False
- for c1, c2 in util.zip_longest(self.constraint.columns, other.columns):
+ for c1, c2 in zip_longest(self.constraint.columns, other.columns):
if (c1 is None and c2 is not None) or (
c2 is None and c1 is not None
):
py38 = sys.version_info >= (3, 8)
py37 = sys.version_info >= (3, 7)
-string_types = (str,)
-binary_type = bytes
-text_type = str
-
# produce a wrapper that allows encoded text to stream
# into a given buffer, but doesn't close it.
from sqlalchemy.util import unique_list # noqa
from .compat import inspect_getfullargspec
-from .compat import string_types
_T = TypeVar("_T")
def to_tuple(x, default=None):
if x is None:
return default
- elif isinstance(x, string_types):
+ elif isinstance(x, str):
return (x,)
elif isinstance(x, Iterable):
return tuple(x)
def dispatch(self, obj: Any, qualifier: str = "default") -> Any:
- if isinstance(obj, string_types):
+ if isinstance(obj, str):
targets: Sequence = [obj]
elif isinstance(obj, type):
targets = obj.__mro__
from sqlalchemy.engine import url
from . import sqla_compat
-from .compat import binary_type
-from .compat import string_types
log = logging.getLogger(__name__)
def write_outstream(stream: TextIO, *text) -> None:
encoding = getattr(stream, "encoding", "ascii") or "ascii"
for t in text:
- if not isinstance(t, binary_type):
+ if not isinstance(t, bytes):
t = t.encode(encoding, "replace")
t = t.decode(encoding)
try:
def format_as_comma(value: Optional[Union[str, "Iterable[str]"]]) -> str:
if value is None:
return ""
- elif isinstance(value, string_types):
+ elif isinstance(value, str):
return value
elif isinstance(value, Iterable):
return ", ".join(value)
from sqlalchemy.sql.elements import TextClause
from sqlalchemy.sql.visitors import traverse
-from . import compat
-
if TYPE_CHECKING:
from sqlalchemy import Index
from sqlalchemy import Table
table: "Table", text_: Union[str, "TextClause", "ColumnElement"]
) -> Union["ColumnElement", "Column"]:
"""a workaround for the Index construct's severe lack of flexibility"""
- if isinstance(text_, compat.string_types):
+ if isinstance(text_, str):
c = Column(text_, sqltypes.NULLTYPE)
table.append_column(c)
return c
from alembic.testing.fixtures import capture_context_buffer
from alembic.testing.fixtures import FutureEngineMixin
from alembic.testing.fixtures import TestBase
-from alembic.util import compat
class PatchEnvironment:
assert isinstance(step.is_upgrade, bool)
assert isinstance(step.is_stamp, bool)
assert isinstance(step.is_migration, bool)
- assert isinstance(step.up_revision_id, compat.string_types)
+ assert isinstance(step.up_revision_id, str)
assert isinstance(step.up_revision, Script)
for revtype in "up", "down", "source", "destination":
assert isinstance(rev, Script)
revids = getattr(step, "%s_revision_ids" % revtype)
for revid in revids:
- assert isinstance(revid, compat.string_types)
+ assert isinstance(revid, str)
heads = kw["heads"]
assert hasattr(heads, "__iter__")
for h in heads:
- assert h is None or isinstance(h, compat.string_types)
+ assert h is None or isinstance(h, str)
class OfflineTransactionalDDLTest(TestBase):
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
