- id: flake8
additional_dependencies:
- flake8-import-order
+ - flake8-import-single==0.1.5
- flake8-builtins
- flake8-future-annotations>=0.0.5
- flake8-docstrings>=1.6.0
"""
+import datetime
+
+from sqlalchemy import Column
+from sqlalchemy import create_engine
+from sqlalchemy import DateTime
from sqlalchemy import event
+from sqlalchemy import Integer
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import Session
def configure_listener(mapper, class_):
if __name__ == "__main__":
- from sqlalchemy import Column, Integer, DateTime, create_engine
- from sqlalchemy.orm import Session
- from sqlalchemy.ext.declarative import declarative_base
- import datetime
-
Base = declarative_base()
event.listen(Base, "mapper_configured", configure_listener, propagate=True)
"""
+from sqlalchemy import Column
from sqlalchemy import event
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship
def configure_listener(class_, key, inst):
if __name__ == "__main__":
- from sqlalchemy import Column, Integer, String, ForeignKey
- from sqlalchemy.orm import relationship
- from sqlalchemy.ext.declarative import declarative_base
-
class Base:
def receive_change_event(self, verb, key, value, oldvalue):
s = "Value '%s' %s on attribute '%s', " % (value, verb, key)
"""
+from dogpile.cache import make_region
from dogpile.cache.api import CacheBackend
from dogpile.cache.api import NO_VALUE
from dogpile.cache.region import register_backend
-from examples.dogpile_caching import environment
+
+from . import environment
+from .caching_query import FromCache
+from .environment import regions
+from .environment import Session
class ScopedSessionBackend(CacheBackend):
if __name__ == "__main__":
- from .environment import Session, regions
- from .caching_query import FromCache
- from dogpile.cache import make_region
# set up a region based on the ScopedSessionBackend,
# pointing to the scoped_session declared in the example
from sqlalchemy import Boolean
from sqlalchemy import Column
+from sqlalchemy import create_engine
from sqlalchemy import event
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
from sqlalchemy import orm
+from sqlalchemy import select
+from sqlalchemy import String
from sqlalchemy import true
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
+from sqlalchemy.orm import sessionmaker
@event.listens_for(Session, "do_orm_execute")
if __name__ == "__main__":
- from sqlalchemy import Integer, Column, String, ForeignKey, Boolean
- from sqlalchemy import select
- from sqlalchemy import create_engine
- from sqlalchemy.orm import relationship, sessionmaker
- from sqlalchemy.ext.declarative import declarative_base
-
Base = declarative_base()
class User(HasPrivate, Base):
import datetime
from sqlalchemy import Column
+from sqlalchemy import create_engine
from sqlalchemy import DateTime
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
from sqlalchemy import orm
+from sqlalchemy import select
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import selectinload
+from sqlalchemy.orm import sessionmaker
class HasTemporal:
if __name__ == "__main__":
- from sqlalchemy import Integer, Column, ForeignKey
- from sqlalchemy import select
- from sqlalchemy import create_engine
- from sqlalchemy.orm import relationship, sessionmaker, selectinload
- from sqlalchemy.ext.declarative import declarative_base
-
Base = declarative_base()
class Parent(HasTemporal, Base):
"""
+from sqlalchemy import and_
+from sqlalchemy import Boolean
+from sqlalchemy import case
+from sqlalchemy import cast
+from sqlalchemy import Column
+from sqlalchemy import create_engine
from sqlalchemy import event
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
from sqlalchemy import literal_column
+from sqlalchemy import null
+from sqlalchemy import or_
+from sqlalchemy import String
+from sqlalchemy import Unicode
+from sqlalchemy import UnicodeText
+from sqlalchemy.ext.associationproxy import association_proxy
+from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.hybrid import hybrid_property
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import Session
+from sqlalchemy.orm.collections import attribute_keyed_dict
from sqlalchemy.orm.interfaces import PropComparator
from .dictlike import ProxiedDictMixin
if __name__ == "__main__":
- from sqlalchemy import (
- Column,
- Integer,
- Unicode,
- ForeignKey,
- UnicodeText,
- and_,
- or_,
- String,
- Boolean,
- cast,
- null,
- case,
- create_engine,
- )
- from sqlalchemy.orm import relationship, Session
- from sqlalchemy.orm.collections import attribute_keyed_dict
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.ext.associationproxy import association_proxy
Base = declarative_base()
"""
+from sqlalchemy import and_
+from sqlalchemy import Column
+from sqlalchemy import create_engine
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import Unicode
+from sqlalchemy import UnicodeText
+from sqlalchemy.ext.associationproxy import association_proxy
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship
+from sqlalchemy.orm import Session
+from sqlalchemy.orm.collections import attribute_keyed_dict
+
class ProxiedDictMixin:
"""Adds obj[key] access to a mapped class.
if __name__ == "__main__":
- from sqlalchemy import (
- Column,
- Integer,
- Unicode,
- ForeignKey,
- UnicodeText,
- and_,
- create_engine,
- )
- from sqlalchemy.orm import relationship, Session
- from sqlalchemy.orm.collections import attribute_keyed_dict
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.ext.associationproxy import association_proxy
Base = declarative_base()
if _fallback_dict is None:
import sqlalchemy
- from sqlalchemy.orm import foreign, remote
+ from . import foreign
+ from . import remote
_fallback_dict = util.immutabledict(sqlalchemy.__dict__).union(
{"foreign": foreign, "remote": remote}
from .base import _SET_DEFERRED_EXPIRED
from .base import PassiveFlag
from .context import FromStatement
+from .context import ORMCompileState
+from .context import QueryContext
from .util import _none_set
from .util import state_str
from .. import exc as sa_exc
if TYPE_CHECKING:
from ._typing import _IdentityKeyType
from .base import LoaderCallableStatus
- from .context import QueryContext
from .interfaces import ORMOption
from .mapper import Mapper
from .query import Query
assert not q._is_lambda_element
- # TODO: fix these imports ....
- from .context import QueryContext, ORMCompileState
-
if load_options is None:
load_options = QueryContext.default_load_options
from .entities import ComparableMixin # noqa
from .util import adict
from .util import drop_all_tables_from_metadata
+from .. import Column
from .. import event
+from .. import func
+from .. import Integer
+from .. import select
+from .. import Table
from .. import util
from ..orm import DeclarativeBase
from ..orm import events as orm_events
def trans_ctx_manager_fixture(self, request, metadata):
rollback, second_operation, begin_nested = request.param
- from sqlalchemy import Table, Column, Integer, func, select
- from . import eq_
-
t = Table("test", metadata, Column("data", Integer))
eng = getattr(self, "bind", None) or config.db
import weakref
import sqlalchemy as sa
+from sqlalchemy import and_
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import inspect
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
+from sqlalchemy import types
from sqlalchemy import Unicode
from sqlalchemy import util
+from sqlalchemy.dialects import mysql
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.dialects import sqlite
from sqlalchemy.engine import result
from sqlalchemy.engine.processors import to_decimal_processor_factory
from sqlalchemy.orm import aliased
from sqlalchemy.orm.session import _sessions
from sqlalchemy.sql import column
from sqlalchemy.sql import util as sql_util
+from sqlalchemy.sql.util import visit_binary_product
from sqlalchemy.sql.visitors import cloned_traverse
from sqlalchemy.sql.visitors import replacement_traverse
from sqlalchemy.testing import engines
"""test storage of bind processors, result processors
in dialect-wide registry."""
- from sqlalchemy.dialects import mysql, postgresql, sqlite
- from sqlalchemy import types
-
eng = engines.testing_engine()
for args in (
(types.Integer, {}),
def test_visit_binary_product(self):
a, b, q, e, f, j, r = (column(chr_) for chr_ in "abqefjr")
- from sqlalchemy import and_, func
- from sqlalchemy.sql.util import visit_binary_product
-
expr = and_((a + b) == q + func.sum(e + f), j == r)
def visit(expr, left, right):
from sqlalchemy.dialects import postgresql
from sqlalchemy.dialects.postgresql import aggregate_order_by
from sqlalchemy.dialects.postgresql import array
+from sqlalchemy.dialects.postgresql import array_agg
+from sqlalchemy.dialects.postgresql import base
from sqlalchemy.dialects.postgresql import DATEMULTIRANGE
from sqlalchemy.dialects.postgresql import DATERANGE
from sqlalchemy.dialects.postgresql import DOMAIN
from sqlalchemy.dialects.postgresql import NamedType
from sqlalchemy.dialects.postgresql import NUMMULTIRANGE
from sqlalchemy.dialects.postgresql import NUMRANGE
+from sqlalchemy.dialects.postgresql import pg8000
+from sqlalchemy.dialects.postgresql import psycopg2
+from sqlalchemy.dialects.postgresql import psycopg2cffi
from sqlalchemy.dialects.postgresql import Range
from sqlalchemy.dialects.postgresql import TSMULTIRANGE
from sqlalchemy.dialects.postgresql import TSRANGE
__backend__ = True
def test_numeric_codes(self):
- from sqlalchemy.dialects.postgresql import (
- base,
- pg8000,
- psycopg2,
- psycopg2cffi,
- )
dialects = (
pg8000.dialect(),
argnames="with_enum, using_aggregate_order_by",
)
def test_array_agg_specific(self, with_enum, using_aggregate_order_by):
- from sqlalchemy.dialects.postgresql import (
- ENUM,
- aggregate_order_by,
- array_agg,
- )
element = ENUM(name="pgenum") if with_enum else Integer()
element_type = type(element)
def async_trans_ctx_manager_fixture(self, request, metadata):
rollback, run_second_execute, begin_nested = request.param
- from sqlalchemy import Table, Column, Integer, func, select
-
t = Table("test", metadata, Column("data", Integer))
eng = getattr(self, "bind", None) or config.db
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import synonym
+from sqlalchemy.sql import coercions
from sqlalchemy.sql import operators
+from sqlalchemy.sql import roles
from sqlalchemy.sql import update
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
def test_expression_isnt_clause_element(self):
A = self._wrong_expr_fixture()
- from sqlalchemy.sql import coercions, roles
-
with testing.expect_raises_message(
exc.InvalidRequestError,
'When interpreting attribute "A.value" as a SQL expression, '
from sqlalchemy.orm.decl_base import _DeferredMapperConfig
from sqlalchemy.orm.events import InstrumentationEvents
from sqlalchemy.orm.events import MapperEvents
+from sqlalchemy.schema import PrimaryKeyConstraint
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assertions
eq_(Foo.__table__.name, "foobat")
def test_table_cls_attribute_return_none(self):
- from sqlalchemy.schema import Column, PrimaryKeyConstraint
+ # this is separate from the "fixture" version of Column used in the
+ # rest of this suite
+ from sqlalchemy.schema import Column
class AutoTable:
@declared_attr
def test_loader_criteria(self):
User, Address = self.classes("User", "Address")
- from sqlalchemy import Column, Integer, String
-
class Foo:
id = Column(Integer)
name = Column(String)
)
def test_loader_criteria_bound_param_thing(self):
- from sqlalchemy import Column, Integer
-
class Foo:
id = Column(Integer)
)
# these must be module level for pickling
- from .test_pickled import User, Address, Dingaling
+ from .test_pickled import Address
+ from .test_pickled import Dingaling
+ from .test_pickled import User
self.mapper_registry.map_imperatively(
User,
"""test the inspection registry system."""
+import random
+import textwrap
+
+from sqlalchemy import Column
from sqlalchemy import exc
from sqlalchemy import ForeignKey
from sqlalchemy import inspect
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import Table
from sqlalchemy import testing
+from sqlalchemy.ext.associationproxy import association_proxy
+from sqlalchemy.ext.associationproxy import AssociationProxyExtensionType
+from sqlalchemy.ext.hybrid import hybrid_method
+from sqlalchemy.ext.hybrid import hybrid_property
+from sqlalchemy.ext.hybrid import HybridExtensionType
from sqlalchemy.orm import aliased
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import relationship
from sqlalchemy.orm.attributes import instance_state
from sqlalchemy.orm.attributes import NO_VALUE
from sqlalchemy.orm.base import InspectionAttr
+from sqlalchemy.orm.interfaces import NotExtension
from sqlalchemy.orm.util import identity_key
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import eq_
assert hasattr(prop, "expression")
def test_extension_types(self):
- from sqlalchemy.ext.associationproxy import (
- association_proxy,
- AssociationProxyExtensionType,
- )
- from sqlalchemy.ext.hybrid import (
- hybrid_property,
- hybrid_method,
- HybridExtensionType,
- )
- from sqlalchemy import Table, MetaData, Integer, Column
- from sqlalchemy.orm.interfaces import NotExtension
-
class SomeClass(self.classes.User):
some_assoc = association_proxy("addresses", "email_address")
return list(names.difference(keyword.kwlist))
def _ordered_name_fixture(self, glbls, clsname, base, supercls):
- import random
- from sqlalchemy import Integer, Column
- import textwrap
names = self._random_names()
class MyClass:
pass
- from sqlalchemy import Table, MetaData, Column, Integer
-
names = self._random_names()
m = MetaData()
from sqlalchemy.orm import RelationshipProperty
from sqlalchemy.orm import Session
from sqlalchemy.orm import synonym
+from sqlalchemy.orm.base import _is_aliased_class
+from sqlalchemy.orm.base import _is_mapped_class
from sqlalchemy.orm.persistence import _sort_states
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
addresses = self.tables.addresses
Address = self.classes.Address
- from sqlalchemy.orm.base import _is_mapped_class, _is_aliased_class
-
class Foo:
x = "something"
from sqlalchemy.testing.assertions import eq_
from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.pickleable import Address
+from sqlalchemy.testing.pickleable import User
from test.orm import _fixtures
from .inheritance._poly_fixtures import _Polymorphic
from .inheritance._poly_fixtures import Company
@testing.fixture
def user_address_fixture(self, registry):
- from sqlalchemy.testing.pickleable import User, Address
registry.map_imperatively(
User,
from timeit import timeit
from types import MappingProxyType
+from sqlalchemy import bindparam
+from sqlalchemy import column
+
def test_case(fn):
fn.__test_case__ = True
NUMBER = 1000000
def init_objects(self):
- from sqlalchemy import column, bindparam
self.object_1 = column("x")
self.object_2 = bindparam("y")
from sqlalchemy.sql.elements import Grouping
from sqlalchemy.sql.expression import ClauseElement
from sqlalchemy.sql.expression import ClauseList
+from sqlalchemy.sql.expression import ColumnClause
+from sqlalchemy.sql.expression import TableClause
from sqlalchemy.sql.selectable import LABEL_STYLE_NONE
from sqlalchemy.sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
from sqlalchemy.testing import assert_raises
class KwargPropagationTest(fixtures.TestBase):
@classmethod
def setup_test_class(cls):
- from sqlalchemy.sql.expression import ColumnClause, TableClause
-
class CatchCol(ColumnClause):
pass
from sqlalchemy import Sequence
from sqlalchemy import String
from sqlalchemy import testing
+from sqlalchemy.dialects.postgresql import ARRAY
+from sqlalchemy.dialects.postgresql import array
from sqlalchemy.schema import CreateTable
from sqlalchemy.sql import literal_column
from sqlalchemy.sql import select
)
def test_literal_binds_pgarray(self):
- from sqlalchemy.dialects.postgresql import ARRAY, array
m = MetaData()
t = Table(
from sqlalchemy.dialects import oracle
from sqlalchemy.dialects import postgresql
from sqlalchemy.dialects import sqlite
+from sqlalchemy.dialects.postgresql import ARRAY as PG_ARRAY
+from sqlalchemy.dialects.postgresql import array
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql import column
from sqlalchemy.sql import functions
eq_(expr.type.dimensions, col.type.dimensions)
def test_array_agg_array_literal_implicit_type(self):
- from sqlalchemy.dialects.postgresql import array, ARRAY as PG_ARRAY
expr = array([column("data", Integer), column("d2", Integer)])
from sqlalchemy import Unicode
from sqlalchemy import UniqueConstraint
from sqlalchemy.engine import default
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.ext.compiler import deregister
from sqlalchemy.schema import AddConstraint
from sqlalchemy.schema import CreateIndex
from sqlalchemy.schema import DefaultClause
assert c in t.indexes
def test_auto_append_lowercase_table(self):
- from sqlalchemy import table, column
t = table("t", column("a"))
t2 = table("t2", column("a"))
)
def test_custom_create(self):
- from sqlalchemy.ext.compiler import compiles, deregister
-
@compiles(schema.CreateColumn)
def compile_(element, compiler, **kw):
column = element.element
flake8-future-annotations>=0.0.5
flake8-docstrings>=1.6.0
flake8-rst-docstrings
+ flake8-import-single==0.1.5
# flake8-rst-docstrings dependency, leaving it here
# in case it requires a version pin
pydocstyle