additional_dependencies:
- flake8-import-order
- flake8-builtins
- - flake8-future-annotations
+ - flake8-future-annotations>=0.0.5
- flake8-docstrings>=1.6.0
- flake8-rst-docstrings
# flake8-rst-docstrings dependency, leaving it here
from ...types import CHAR
from ...types import CLOB
from ...types import DOUBLE_PRECISION
-from ...types import FLOAT
from ...types import INTEGER
from ...types import NCHAR
from ...types import NVARCHAR
if typing.TYPE_CHECKING:
from types import ModuleType
- from .base import Connection
from .base import Engine
from .interfaces import _CoreMultiExecuteParams
from .interfaces import _CoreSingleExecuteParams
from ..util.typing import Literal
if typing.TYPE_CHECKING:
- from .base import Connection
from .interfaces import _CoreMultiExecuteParams
from .interfaces import _CoreSingleExecuteParams
from .interfaces import _DBAPIAnyExecuteParams
else:
from sqlalchemy.cyextension.processors import (
DecimalResultProcessor,
- ) # noqa
- from sqlalchemy.cyextension.processors import (
+ )
+ from sqlalchemy.cyextension.processors import ( # noqa: F401
int_to_boolean as int_to_boolean,
)
- from sqlalchemy.cyextension.processors import str_to_date as str_to_date
- from sqlalchemy.cyextension.processors import (
+ from sqlalchemy.cyextension.processors import ( # noqa: F401,E501
+ str_to_date as str_to_date,
+ )
+ from sqlalchemy.cyextension.processors import ( # noqa: F401
str_to_datetime as str_to_datetime,
)
- from sqlalchemy.cyextension.processors import str_to_time as str_to_time
- from sqlalchemy.cyextension.processors import to_float as to_float
- from sqlalchemy.cyextension.processors import to_str as to_str
+ from sqlalchemy.cyextension.processors import ( # noqa: F401,E501
+ str_to_time as str_to_time,
+ )
+ from sqlalchemy.cyextension.processors import ( # noqa: F401,E501
+ to_float as to_float,
+ )
+ from sqlalchemy.cyextension.processors import ( # noqa: F401,E501
+ to_str as to_str,
+ )
def to_decimal_processor_factory(target_class, scale):
# Note that the scale argument is not taken into account for integer
from sqlalchemy.cyextension.resultproxy import tuplegetter as tuplegetter
if typing.TYPE_CHECKING:
- from .row import RowMapping
from ..sql.schema import Column
from ..sql.type_api import _ResultProcessorType
from ._py_util import _distill_params_20 as _distill_params_20
from ._py_util import _distill_raw_params as _distill_raw_params
else:
- from sqlalchemy.cyextension.util import (
+ from sqlalchemy.cyextension.util import ( # noqa: F401
_distill_params_20 as _distill_params_20,
)
- from sqlalchemy.cyextension.util import (
+ from sqlalchemy.cyextension.util import ( # noqa: F401
_distill_raw_params as _distill_raw_params,
)
from ...util.typing import Protocol
if TYPE_CHECKING:
- from ...engine import Connection
- from ...engine import Engine
from ...engine.cursor import CursorResult
from ...engine.interfaces import _CoreAnyExecuteParams
from ...engine.interfaces import _CoreSingleExecuteParams
"""
-from ..engine import Connection as Connection
-from ..engine import create_engine as create_engine
-from ..engine import Engine as Engine
+from ..engine import Connection as Connection # noqa: F401
+from ..engine import create_engine as create_engine # noqa: F401
+from ..engine import Engine as Engine # noqa: F401
from . import exc as orm_exc
from . import path_registry
from .base import _MappedAttribute as _MappedAttribute
-from .base import EXT_CONTINUE as EXT_CONTINUE
-from .base import EXT_SKIP as EXT_SKIP
-from .base import EXT_STOP as EXT_STOP
-from .base import InspectionAttr as InspectionAttr
-from .base import InspectionAttrExtensionType as InspectionAttrExtensionType
+from .base import EXT_CONTINUE as EXT_CONTINUE # noqa: F401
+from .base import EXT_SKIP as EXT_SKIP # noqa: F401
+from .base import EXT_STOP as EXT_STOP # noqa: F401
+from .base import InspectionAttr as InspectionAttr # noqa: F401
from .base import InspectionAttrInfo as InspectionAttrInfo
-from .base import MANYTOMANY as MANYTOMANY
-from .base import MANYTOONE as MANYTOONE
-from .base import NotExtension as NotExtension
-from .base import ONETOMANY as ONETOMANY
+from .base import MANYTOMANY as MANYTOMANY # noqa: F401
+from .base import MANYTOONE as MANYTOONE # noqa: F401
+from .base import NotExtension as NotExtension # noqa: F401
+from .base import ONETOMANY as ONETOMANY # noqa: F401
from .base import SQLORMOperations
from .. import ColumnElement
from .. import inspection
if TYPE_CHECKING:
from ._typing import _IdentityKeyType
from .base import LoaderCallableStatus
- from .context import FromStatement
from .interfaces import ORMOption
from .mapper import Mapper
from .session import Session
from .state import InstanceState
from ..engine.interfaces import _ExecuteOptions
from ..sql import Select
- from ..sql.selectable import ForUpdateArg
_T = TypeVar("_T", bound=Any)
_O = TypeVar("_O", bound=object)
from ..sql.elements import ColumnElement
from ..sql.elements import KeyedColumnElement
from ..sql.schema import Column
- from ..sql.schema import Table
from ..sql.selectable import FromClause
from ..sql.util import ColumnAdapter
from ..util import OrderedSet
from ._typing import _EntityType
from .session import Session
from ..engine.result import ScalarResult
- from ..engine.row import Row
from ..sql._typing import _ColumnExpressionArgument
from ..sql._typing import _ColumnsClauseArgument
from ..sql._typing import _MAYBE_ENTITY
@overload
def with_entities(
self, _entity: _EntityType[_O], **kwargs: Any
- ) -> ScalarInstanceQuery[_O]:
+ ) -> Query[_O]:
...
@overload
def and_(
self, *criteria: _ColumnExpressionArgument[bool]
- ) -> interfaces.PropComparator[bool]:
+ ) -> PropComparator[bool]:
"""Add AND criteria.
See :meth:`.PropComparator.and_` for an example.
return _resolver(self.parent.class_, self)
- @util.preload_module("sqlalchemy.orm.mapper")
def _check_conflicts(self):
"""Test that this relationship is legal, warn about
inheritance conflicts."""
- mapperlib = util.preloaded.orm_mapper
if self.parent.non_primary and not class_mapper(
self.parent.class_, configure=False
).has_property(self.key):
from ._typing import _EntityType
from ._typing import _IdentityKeyType
from ._typing import _InstanceDict
- from ._typing import _O
- from .context import FromStatement
from .interfaces import ORMOption
from .interfaces import UserDefinedOption
from .mapper import Mapper
from ..sql.base import Executable
from ..sql.elements import ClauseElement
from ..sql.roles import TypedColumnsClauseRole
- from ..sql.schema import Table
- from ..sql.selectable import Select
from ..sql.selectable import TypedReturnsRows
_T = TypeVar("_T", bound=Any)
from ._typing import _LoaderCallable
from .attributes import AttributeImpl
from .attributes import History
- from .base import LoaderCallableStatus
from .base import PassiveFlag
from .collections import _AdaptedCollectionProtocol
from .identity import IdentityMap
from ._typing import insp_is_mapper
from ._typing import prop_is_relationship
from .base import _class_to_mapper as _class_to_mapper
-from .base import _never_set as _never_set
-from .base import _none_set as _none_set
-from .base import attribute_str as attribute_str
+from .base import _never_set as _never_set # noqa: F401
+from .base import _none_set as _none_set # noqa: F401
+from .base import attribute_str as attribute_str # noqa: F401
from .base import class_mapper as class_mapper
from .base import InspectionAttr as InspectionAttr
-from .base import instance_str as instance_str
+from .base import instance_str as instance_str # noqa: F401
from .base import object_mapper as object_mapper
-from .base import object_state as object_state
+from .base import object_state as object_state # noqa: F401
from .base import opt_manager_of_class
-from .base import state_attribute_str as state_attribute_str
-from .base import state_class_str as state_class_str
-from .base import state_str as state_str
+from .base import state_attribute_str as state_attribute_str # noqa: F401
+from .base import state_class_str as state_class_str # noqa: F401
+from .base import state_str as state_str # noqa: F401
from .interfaces import CriteriaOption
from .interfaces import MapperProperty as MapperProperty
from .interfaces import ORMColumnsClauseRole
from .interfaces import ORMEntityColumnsClauseRole
from .interfaces import ORMFromClauseRole
-from .interfaces import PropComparator as PropComparator
from .path_registry import PathRegistry as PathRegistry
from .. import event
from .. import exc as sa_exc
from ._typing import _ColumnExpressionOrLiteralArgument
from ._typing import _TypeEngineArgument
from .elements import BinaryExpression
- from .functions import FunctionElement
from .selectable import FromClause
from .type_api import TypeEngine
from .selectable import Values
if TYPE_CHECKING:
- from ._typing import _ColumnsClauseArgument
from ._typing import _FromClauseArgument
from ._typing import _OnClauseArgument
from ._typing import _SelectStatementForCompoundArgument
from .dml import ValuesBase
from .elements import ColumnElement
from .schema import _SQLExprDefault
- from .selectable import TableClause
REQUIRED = util.symbol(
"REQUIRED",
from ._typing import _TypedColumnClauseArgument as _TCCA # noqa
from .base import ReadOnlyColumnCollection
from .compiler import SQLCompiler
- from .elements import ColumnElement
from .elements import KeyedColumnElement
from .selectable import _ColumnsClauseElement
from .selectable import _SelectIterable
from ..util.typing import Self
if TYPE_CHECKING:
- from .cache_key import CacheConst
from .elements import BindParameter
from .elements import ClauseElement
from .roles import SQLRole
from ._typing import _InfoType
from ._typing import _TextCoercedExpressionArgument
from ._typing import _TypeEngineArgument
- from .base import DedupeColumnCollection
from .base import ReadOnlyColumnCollection
from .compiler import DDLCompiler
from .elements import BindParameter
:func:`_sql.select` function.
"""
- things = [
- coercions.expect(
- roles.ColumnsClauseRole, ent, apply_propagate_attrs=self
- )
- for ent in entities
- ]
-
self._raw_columns = [
coercions.expect(
roles.ColumnsClauseRole, ent, apply_propagate_attrs=self
from .elements import ColumnElement
from .operators import OperatorType
from .sqltypes import _resolve_value_to_type as _resolve_value_to_type
- from .sqltypes import BOOLEANTYPE as BOOLEANTYPE
- from .sqltypes import INDEXABLE as INDEXABLE
- from .sqltypes import INTEGERTYPE as INTEGERTYPE
- from .sqltypes import MATCHTYPE as MATCHTYPE
+ from .sqltypes import BOOLEANTYPE as BOOLEANTYPE # noqa: F401
+ from .sqltypes import INDEXABLE as INDEXABLE # noqa: F401
+ from .sqltypes import INTEGERTYPE as INTEGERTYPE # noqa: F401
+ from .sqltypes import MATCHTYPE as MATCHTYPE # noqa: F401
from .sqltypes import NULLTYPE as NULLTYPE
- from .sqltypes import STRINGTYPE as STRINGTYPE
- from .sqltypes import TABLEVALUE as TABLEVALUE
+ from .sqltypes import STRINGTYPE as STRINGTYPE # noqa: F401
+ from .sqltypes import TABLEVALUE as TABLEVALUE # noqa: F401
from ..engine.interfaces import Dialect
from . import roles
from . import visitors
from ._typing import is_text_clause
-from .annotation import _deep_annotate as _deep_annotate
-from .annotation import _deep_deannotate as _deep_deannotate
-from .annotation import _shallow_annotate as _shallow_annotate
+from .annotation import _deep_annotate as _deep_annotate # noqa: F401
+from .annotation import _deep_deannotate as _deep_deannotate # noqa: F401
+from .annotation import _shallow_annotate as _shallow_annotate # noqa: F401
from .base import _expand_cloned
from .base import _from_objects
-from .cache_key import HasCacheKey as HasCacheKey
-from .ddl import sort_tables as sort_tables
+from .cache_key import HasCacheKey as HasCacheKey # noqa: F401
+from .ddl import sort_tables as sort_tables # noqa: F401
from .elements import _find_columns as _find_columns
from .elements import _label_reference
from .elements import _textual_label_reference
from ._py_util import prefix_anon_map as prefix_anon_map
from ._py_util import cache_anon_map as anon_map
else:
- from sqlalchemy.cyextension.util import prefix_anon_map as prefix_anon_map
- from sqlalchemy.cyextension.util import cache_anon_map as anon_map
+ from sqlalchemy.cyextension.util import ( # noqa: F401,E501
+ prefix_anon_map as prefix_anon_map,
+ )
+ from sqlalchemy.cyextension.util import ( # noqa: F401,E501
+ cache_anon_map as anon_map,
+ )
__all__ = [
else ""
)
- def is_exit_exception(e):
+ def is_exit_exception(e): # noqa: F811
return not isinstance(e, Exception)
- def await_only(thing): # type: ignore
+ def await_only(thing): # type: ignore # noqa: F811
_not_implemented()
- def await_fallback(thing): # type: ignore
+ def await_fallback(thing): # type: ignore # noqa: F811
return thing
- def greenlet_spawn(fn, *args, **kw): # type: ignore
+ def greenlet_spawn(fn, *args, **kw): # type: ignore # noqa: F811
_not_implemented()
- def AsyncAdaptedLock(*args, **kw): # type: ignore
+ def AsyncAdaptedLock(*args, **kw): # type: ignore # noqa: F811
_not_implemented()
- def _util_async_run(fn, *arg, **kw): # type: ignore
+ def _util_async_run(fn, *arg, **kw): # type: ignore # noqa: F811
return fn(*arg, **kw)
- def _util_async_run_coroutine_function(fn, *arg, **kw): # type: ignore
+ def _util_async_run_coroutine_function(fn, *arg, **kw): # type: ignore # noqa: F811,E501
_not_implemented()
from typing import TypeVar
from typing import Union
-from typing_extensions import NotRequired as NotRequired
+from typing_extensions import NotRequired as NotRequired # noqa: F401
from . import compat
from typing import TypeGuard as TypeGuard
from typing import TypeAlias as TypeAlias
else:
- from typing_extensions import TypeGuard as TypeGuard
- from typing_extensions import TypeAlias as TypeAlias
+ from typing_extensions import TypeGuard as TypeGuard # noqa: F401
+ from typing_extensions import TypeAlias as TypeAlias # noqa: F401
if typing.TYPE_CHECKING or compat.py38:
from typing import SupportsIndex as SupportsIndex
else:
- from typing_extensions import SupportsIndex as SupportsIndex
+ from typing_extensions import SupportsIndex as SupportsIndex # noqa: F401
if typing.TYPE_CHECKING or compat.py310:
from typing import Annotated as Annotated
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_not
from sqlalchemy.testing import is_true
-from sqlalchemy.testing import mock
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
# for aliased() to return some namespace of User that's not User.
# AsAliased superclass type was tested for this but it had its own
# awkwardnesses that aren't really worth it
- x = ua(id=1, name="foo")
+ ua(id=1, name="foo")
# EXPECTED_TYPE: Type[User]
reveal_type(ua)
__sa_dataclass_metadata_key__ = "sa"
id: Mapped[int] = mapped_column(primary_key=True)
- bs: List["B"] = dataclasses.field(
+ bs: List["B"] = dataclasses.field( # noqa: F821
default_factory=list, metadata={"sa": relationship()}
)
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs: Mapped[List["B"]] = relationship(back_populates="a")
+ bs: Mapped[List["B"]] = relationship( # noqa: F821
+ back_populates="a"
+ )
class B(decl_base):
__tablename__ = "b"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs_list: Mapped[List["B"]] = relationship(viewonly=True)
- bs_set: Mapped[Set["B"]] = relationship(viewonly=True)
- bs_list_warg: Mapped[List["B"]] = relationship("B", viewonly=True)
- bs_set_warg: Mapped[Set["B"]] = relationship("B", viewonly=True)
+ bs_list: Mapped[List["B"]] = relationship( # noqa: F821
+ viewonly=True
+ )
+ bs_set: Mapped[Set["B"]] = relationship( # noqa: F821
+ viewonly=True
+ )
+ bs_list_warg: Mapped[List["B"]] = relationship( # noqa: F821
+ "B", viewonly=True
+ )
+ bs_set_warg: Mapped[Set["B"]] = relationship( # noqa: F821
+ "B", viewonly=True
+ )
class B(decl_base):
__tablename__ = "b"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs: Mapped[Dict[str, "B"]] = relationship()
+ bs: Mapped[Dict[str, "B"]] = relationship() # noqa: F821
class B(decl_base):
__tablename__ = "b"
id: Mapped[int] = mapped_column(primary_key=True)
data: Mapped[str] = mapped_column()
- bs: Mapped[MappedCollection[str, "B"]] = relationship(
+ bs: Mapped[
+ MappedCollection[str, "B"] # noqa: F821
+ ] = relationship(
collection_class=attribute_mapped_collection("name")
)
name: Mapped[str50]
- employees: Mapped[Set["Person"]] = relationship()
+ employees: Mapped[Set["Person"]] = relationship() # noqa: F821
class Person(decl_base):
__tablename__ = "person"
t1t2_join = select(t1.c.x).select_from(t1.join(t2)).alias()
def go():
- t1t2_join_2 = select(t1.c.q).select_from(t1.join(t2)).alias()
+ t1t2_join_2 = ( # noqa: F841
+ select(t1.c.q).select_from(t1.join(t2)).alias()
+ )
self.mapper_registry.map_imperatively(
Parent,
t2,
assert isinstance(row, collections_abc.Sequence)
assert isinstance(row._mapping, collections_abc.Mapping)
- def test_multiple_entity_true(self):
+ def test_multiple_entity_true_tuples(self):
User = self.classes.User
query = fixture_session().query(User.id, User).tuples()
is_false(query.is_single_entity)
found_during_type_annotation
"""
- User, Address = self.classes.User, self.classes.Address
+ User = self.classes.User
sess = fixture_session()
u1 = sess.get(User, 7)
r"DefaultGenerator object is deprecated; please use "
r"the .scalar\(\) method."
):
- x = connection.execute(t.c.col1.default)
+ connection.execute(t.c.col1.default)
with expect_deprecated(
r"Using the .execute\(\) method to invoke a "
r"DefaultGenerator object is deprecated; please use "
r"the .scalar\(\) method."
):
- y = connection.execute(t.c.col2.default)
+ connection.execute(t.c.col2.default)
with expect_deprecated(
r"Using the .execute\(\) method to invoke a "
r"DefaultGenerator object is deprecated; please use "
r"the .scalar\(\) method."
):
- z = connection.execute(t.c.col3.default)
+ connection.execute(t.c.col3.default)
def test_standalone_default_scalar(self, connection):
t = self.tables.default_test
expr = op(expr, column(f"d{i}"))
opstring = compiler.OPERATORS[op]
- exprs = opstring.join(f"d{i}" for i in range(num))
if negate:
negate_op = {
flake8
flake8-import-order
flake8-builtins
- flake8-future-annotations
+ flake8-future-annotations>=0.0.5
flake8-docstrings>=1.6.0
flake8-rst-docstrings
# flake8-rst-docstrings dependency, leaving it here
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
