because evaluating the query as a list would incur two SQL calls instead of
one::
- class Iterates(object):
+ class Iterates:
def __len__(self):
print("LEN!")
return 5
.. sourcecode:: python+sql
- class ListLike(object):
+ class ListLike:
def __init__(self):
self.data = []
def append(self, item):
explicit about the interface you are implementing by providing an
``__emulates__`` class attribute::
- class SetLike(object):
+ class SetLike:
__emulates__ = set
def __init__(self):
from sqlalchemy.orm.collections import collection
- class SetLike(object):
+ class SetLike:
__emulates__ = set
def __init__(self):
A simple example represents pairs of columns as a ``Point`` object.
``Point`` represents such a pair as ``.x`` and ``.y``::
- class Point(object):
+ class Point:
def __init__(self, x, y):
self.x = x
self.y = y
from sqlalchemy.orm import composite
- class Point(object):
+ class Point:
def __init__(self, x, y):
self.x = x
self.y = y
def __ne__(self, other):
return not self.__eq__(other)
- class Vertex(object):
+ class Vertex:
def __init__(self, start, end):
self.start = start
self.end = end
from sqlalchemy import orm
- class MyMappedClass(object):
+ class MyMappedClass:
def __init__(self, data):
self.data = data
# we need stuff on all instances, but not in the database.
This is a very open-ended hook that can allow special customizations
to a :class:`_schema.Table` that one generates here::
- class MyMixin(object):
+ class MyMixin:
@classmethod
def __table_cls__(cls, name, metadata_obj, *arg, **kw):
return Table(
inheritance should take place based on the arguments for the table itself,
such as, define as single-inheritance if there is no primary key present::
- class AutoTable(object):
+ class AutoTable:
@declared_attr
def __tablename__(cls):
return cls.__name__
fully "classical" style, without using Declarative at all.
A constructor similar to that supplied by Declarative is illustrated::
- class Employee(object):
+ class Employee:
def __init__(self, **kw):
for k in kw:
setattr(self, k, kw[k])
Column('id', Integer, primary_key=True),
Column('label', String)
)
- class Node(object):
+ class Node:
pass
mapper_registry.map_imperatively(Node, node, properties={
using the :class:`_orm.declared_attr` decorator, such as in the
example at :ref:`orm_declarative_mixins_relationships`::
- class RefTargetMixin(object):
+ class RefTargetMixin:
@declared_attr
def target_id(cls):
return Column('target_id', ForeignKey('target.id'))
### this is the **wrong way to do it** ###
- class ThingOne(object):
+ class ThingOne:
def go(self):
session = Session()
try:
session.rollback()
raise
- class ThingTwo(object):
+ class ThingTwo:
def go(self):
session = Session()
try:
### this is a **better** (but not the only) way to do it ###
- class ThingOne(object):
+ class ThingOne:
def go(self, session):
session.query(FooBar).update({"x": 5})
- class ThingTwo(object):
+ class ThingTwo:
def go(self, session):
session.query(Widget).update({"q": 18})
import datetime
- class HasTimestamp(object):
+ class HasTimestamp:
timestamp = Column(DateTime, default=datetime.datetime.now)
from sqlalchemy.orm.collections import MappedCollection
-class Base(object):
+class Base:
id = Column(Integer, primary_key=True)
return find
-class MyClass(object):
+class MyClass:
__sa_instrumentation_manager__ = MyClassState
def __init__(self, **kwargs):
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
- class Base(object):
+ class Base:
def receive_change_event(self, verb, key, value, oldvalue):
s = "Value '%s' %s on attribute '%s', " % (value, verb, key)
if oldvalue:
from sqlalchemy.orm.interfaces import UserDefinedOption
-class ORMCache(object):
+class ORMCache:
"""An add-on for an ORM :class:`.Session` optionally loads full results
from a dogpile cache region.
from sqlalchemy.orm import sessionmaker
-class ProxyDict(object):
+class ProxyDict:
def __init__(self, parent, collection_name, childclass, keyname):
self.parent = parent
self.collection_name = collection_name
# and the ElementTree root element.
-class Document(object):
+class Document:
def __init__(self, name, element):
self.filename = name
self.element = element
# may be backed by native implementations. so here we construct an adapter.
-class _Node(object):
+class _Node:
pass
# stored for a particular Node.
-class _Attribute(object):
+class _Attribute:
def __init__(self, name, value):
self.name = name
self.value = value
# collection.
-class ElementTreeMarshal(object):
+class ElementTreeMarshal:
def __get__(self, document, owner):
if document is None:
return self
# and the ElementTree root element.
-class Document(object):
+class Document:
def __init__(self, name, element):
self.filename = name
self.element = element
# may be backed by native implementations. so here we construct an adapter.
-class _Node(object):
+class _Node:
pass
# stored for a particular Node.
-class _Attribute(object):
+class _Attribute:
def __init__(self, name, value):
self.name = name
self.value = value
# collection.
-class ElementTreeMarshal(object):
+class ElementTreeMarshal:
def __get__(self, document, owner):
if document is None:
return self
# and the ElementTree root element.
-class Document(object):
+class Document:
def __init__(self, name, element):
self.filename = name
self.element = element
)
-class HasPrivate(object):
+class HasPrivate:
"""Mixin that identifies a class as having private entities"""
public = Column(Boolean, nullable=False)
from sqlalchemy import orm
-class HasTemporal(object):
+class HasTemporal:
"""Mixin that identifies a class as having a timestamp column"""
timestamp = Column(
@as_declarative()
-class Base(object):
+class Base:
"""Base class which provides automated table name
and surrogate primary key column.
)
-class HasAddresses(object):
+class HasAddresses:
"""HasAddresses mixin, creates a relationship to
the address_association table for each parent.
@as_declarative()
-class Base(object):
+class Base:
"""Base class which provides automated table name
and surrogate primary key column.
)
-class HasAddresses(object):
+class HasAddresses:
"""HasAddresses mixin, creates a relationship to
the address_association table for each parent.
@as_declarative()
-class Base(object):
+class Base:
"""Base class which provides automated table name
and surrogate primary key column.
)
-class HasAddresses(object):
+class HasAddresses:
"""HasAddresses mixin, creates a new address_association
table for each parent.
@as_declarative()
-class Base(object):
+class Base:
"""Base class which provides automated table name
and surrogate primary key column.
id = Column(Integer, primary_key=True)
-class Address(object):
+class Address:
"""Define columns that will be present in each
'Address' table.
)
-class HasAddresses(object):
+class HasAddresses:
"""HasAddresses mixin, creates a new Address class
for each parent.
)
-class Organization(object):
+class Organization:
def __init__(self, name):
self.name = name
-class Member(object):
+class Member:
def __init__(self, name):
self.name = name
import time
-class Profiler(object):
+class Profiler:
tests = []
_setup = None
return suites
-class TestResult(object):
+class TestResult:
def __init__(
self, profile, test, stats=None, total_time=None, sort="cumulative"
):
# because if you're going to roll your own, you're probably
# going to do this, so see how this pushes you right back into
# ORM land anyway :)
- class SimpleCustomer(object):
+ class SimpleCustomer:
def __init__(self, id_, name, description):
self.id_ = id_
self.name = name
# Python datatypes
-class GisElement(object):
+class GisElement:
"""Represents a geometry value."""
def __str__(self):
local_mapper.version_id_col = local_mapper.local_table.c.version
-class Versioned(object):
+class Versioned:
use_mapper_versioning = False
"""if True, also assign the version column to be tracked by the mapper"""
from sqlalchemy.orm import sessionmaker
-class Versioned(object):
+class Versioned:
def new_version(self, session):
# make us transient (removes persistent
# identity).
from sqlalchemy.orm import sessionmaker
-class Versioned(object):
+class Versioned:
# we have a composite primary key consisting of "id"
# and "version_id"
id = Column(Integer, primary_key=True)
return now
-class VersionedStartEnd(object):
+class VersionedStartEnd:
start = Column(DateTime, primary_key=True)
end = Column(DateTime, primary_key=True)
from .dictlike import ProxiedDictMixin
-class PolymorphicVerticalProperty(object):
+class PolymorphicVerticalProperty:
"""A key/value pair with polymorphic value storage.
The class which is mapped should indicate typing information
from __future__ import unicode_literals
-class ProxiedDictMixin(object):
+class ProxiedDictMixin:
"""Adds obj[key] access to a mapped class.
This class basically proxies dictionary access to an attribute
# the MIT License: https://www.opensource.org/licenses/mit-license.php
-class Connector(object):
+class Connector:
pass
__visit_name__ = "_BASETIMEIMPL"
-class _DateTimeBase(object):
+class _DateTimeBase:
def bind_processor(self, dialect):
def process(value):
if type(value) == datetime.date:
self.precision = precision
-class _UnicodeLiteral(object):
+class _UnicodeLiteral:
def literal_processor(self, dialect):
def process(value):
# Note: these objects currently match exactly those of MySQL, however since
# these are not generalizable to all JSON implementations, remain separately
# implemented for each dialect.
-class _FormatTypeMixin(object):
+class _FormatTypeMixin:
def _format_value(self, value):
raise NotImplementedError()
from ...connectors.pyodbc import PyODBCConnector
-class _ms_numeric_pyodbc(object):
+class _ms_numeric_pyodbc:
"""Turns Decimals with adjusted() < 0 or > 7 into strings.
pass
-class _ms_binary_pyodbc(object):
+class _ms_binary_pyodbc:
"""Wraps binary values in dialect-specific Binary wrapper.
If the value is null, return a pyodbc-specific BinaryNull
object to prevent pyODBC [and FreeTDS] from defaulting binary
return process
-class _ODBCDateTimeBindProcessor(object):
+class _ODBCDateTimeBindProcessor:
"""Add bind processors to handle datetimeoffset behaviors"""
has_tz = False
return rows
-class _DecodingRow(object):
+class _DecodingRow:
"""Return unicode-decoded values based on type inspection.
Smooth over data type issues (esp. with alpha driver versions) and
pass
-class _FormatTypeMixin(object):
+class _FormatTypeMixin:
def _format_value(self, value):
raise NotImplementedError()
from ... import util
-class ReflectedState(object):
+class ReflectedState:
"""Stores raw information about a SHOW CREATE TABLE statement."""
def __init__(self):
@log.class_logger
-class MySQLTableDefinitionParser(object):
+class MySQLTableDefinitionParser:
"""Parses the results of a SHOW CREATE TABLE statement."""
def __init__(self, dialect, preparer):
from ... import util
-class _NumericType(object):
+class _NumericType:
"""Base for MySQL numeric types.
This is the base both for NUMERIC as well as INTEGER, hence
__all__ = ("INT4RANGE", "INT8RANGE", "NUMRANGE")
-class RangeOperators(object):
+class RangeOperators:
"""
This mixin provides functionality for the Range Operators
listed in Table 9-44 of the `PostgreSQL documentation`__ for Range
return process
-class _DateTimeMixin(object):
+class _DateTimeMixin:
_reg = None
_storage_format = None
# Note: these objects currently match exactly those of MySQL, however since
# these are not generalizable to all JSON implementations, remain separately
# implemented for each dialect.
-class _FormatTypeMixin(object):
+class _FormatTypeMixin:
def _format_value(self, value):
raise NotImplementedError()
return self.pool.connect()
-class OptionEngineMixin(object):
+class OptionEngineMixin:
_sa_propagate_class_events = False
def __init__(self, proxied, execution_options):
self._translated_indexes = self._tuplefilter = None
-class ResultFetchStrategy(object):
+class ResultFetchStrategy:
"""Define a fetching strategy for a result object.
_NO_RESULT_METADATA = _NoResultMetaData()
-class BaseCursorResult(object):
+class BaseCursorResult:
"""Base class for database result objects."""
out_parameters = None
return connection
-class _RendersLiteral(object):
+class _RendersLiteral:
def literal_processor(self, dialect):
def process(value):
return "'%s'" % value
from ..sql.compiler import TypeCompiler # noqa
-class Dialect(object):
+class Dialect:
"""Define the behavior of a specific database and DB-API combination.
Any aspect of metadata definition, SQL query generation,
raise NotImplementedError()
-class CreateEnginePlugin(object):
+class CreateEnginePlugin:
"""A set of hooks intended to augment the construction of an
:class:`_engine.Engine` object based on entrypoint names in a URL.
"""
-class ExecutionContext(object):
+class ExecutionContext:
"""A messenger object for a Dialect that corresponds to a single
execution.
raise NotImplementedError()
-class ExceptionContext(object):
+class ExceptionContext:
"""Encapsulate information about an error condition in progress.
This object exists solely to be passed to the
"""
-class AdaptedConnection(object):
+class AdaptedConnection:
"""Interface of an adapted connection object to support the DBAPI protocol.
Used by asyncio dialects to provide a sync-style pep-249 facade on top
@inspection._self_inspects
-class Inspector(object):
+class Inspector:
"""Performs database schema inspection.
The Inspector acts as a proxy to the reflection methods of the
return lambda row: (it(row),)
-class ResultMetaData(object):
+class ResultMetaData:
"""Base for metadata about result rows."""
__slots__ = ()
return uniques, strategy
-class _WithKeys(object):
+class _WithKeys:
# used mainly to share documentation on the keys method.
def keys(self):
"""Return an iterable view which yields the string keys that would
)
-class FrozenResult(object):
+class FrozenResult:
"""Represents a :class:`.Result` object in a "frozen" state suitable
for caching.
except ImportError:
_baserow_usecext = False
- class BaseRow(object):
+ class BaseRow:
__slots__ = ("_parent", "_data", "_keymap", "_key_style")
def __init__(self, parent, processors, keymap, key_style, data):
from .mock import MockConnection # noqa
-class MockEngineStrategy(object):
+class MockEngineStrategy:
MockConnection = MockConnection
raise exc.ArgumentError("mapping or sequence expected for parameters")
-class TransactionalContext(object):
+class TransactionalContext:
"""Apply Python context manager behavior to transaction objects.
Performs validation to ensure the subject of the transaction is not
return weakref.ref(self, registry._collection_gced)
-class _empty_collection(object):
+class _empty_collection:
def append(self, element):
pass
) or name.startswith("_sa_event")
-class _UnpickleDispatch(object):
+class _UnpickleDispatch:
"""Serializable callable that re-generates an instance of
:class:`_Dispatch` given a particular :class:`.Events` subclass.
raise AttributeError("No class with a 'dispatch' member present.")
-class _Dispatch(object):
+class _Dispatch:
"""Mirror the event listening definitions of an Events class with
listener collections.
cls.dispatch._clear()
-class _JoinedDispatcher(object):
+class _JoinedDispatcher:
"""Represent a connection between two _Dispatch objects."""
__slots__ = "local", "parent", "_instance_cls"
return self.parent._events
-class dispatcher(object):
+class dispatcher:
"""Descriptor used by target classes to
deliver the _Dispatch class at the class level
and produce new _Dispatch instances for target
del _key_to_collection[key]
-class _EventKey(object):
+class _EventKey:
"""Represent :func:`.listen` arguments."""
__slots__ = (
_version_token = None
-class HasDescriptionCode(object):
+class HasDescriptionCode:
"""helper which adds 'code' as an attribute and '_code_str' as a method"""
code = None
"""SQL was attempted without a database connection to execute it on."""
-class DontWrapMixin(object):
+class DontWrapMixin:
"""A mixin class which, when applied to a user-defined Exception class,
will not be wrapped inside of :exc:`.StatementError` if the error is
emitted within the process of executing a statement.
)
-class AssociationProxyInstance(object):
+class AssociationProxyInstance:
"""A per-class object that serves class- and object-specific results.
This is used by :class:`.AssociationProxy` when it is invoked
)
-class _lazy_collection(object):
+class _lazy_collection:
def __init__(self, obj, target):
self.parent = obj
self.target = target
self.target = state["target"]
-class _AssociationCollection(object):
+class _AssociationCollection:
def __init__(self, lazy_collection, creator, getter, setter, parent):
"""Constructs an _AssociationCollection.
raise TypeError("Unknown relationship function: %s" % return_fn)
-class AutomapBase(object):
+class AutomapBase:
"""Base class for an "automap" schema.
The :class:`.AutomapBase` class can be compared to the "declarative base"
log = logging.getLogger(__name__)
-class Bakery(object):
+class Bakery:
"""Callable which returns a :class:`.BakedQuery`.
This object is returned by the class method
return self.cls(self.cache, initial_fn, args)
-class BakedQuery(object):
+class BakedQuery:
"""A builder object for :class:`.query.Query` objects."""
__slots__ = "steps", "_bakery", "_cache_key", "_spoiled"
return query
-class Result(object):
+class Result:
"""Invokes a :class:`.BakedQuery` against a :class:`.Session`.
The :class:`_baked.Result` object is where the actual :class:`.query.Query`
del class_._compiler_dispatcher
-class _dispatcher(object):
+class _dispatcher:
def __init__(self):
self.specs = {}
)
-class ConcreteBase(object):
+class ConcreteBase:
"""A helper class for 'concrete' declarative mappings.
:class:`.ConcreteBase` will use the :func:`.polymorphic_union`
)
-class DeferredReflection(object):
+class DeferredReflection:
"""A helper class for construction of mappings based on
a deferred reflection step.
from sqlalchemy import func
- class Interval(object):
+ class Interval:
# ...
@hybrid_property
the new hybrid with the additional state will be attached to the class
with the non-matching name. To use the example above::
- class Interval(object):
+ class Interval:
# ...
@hybrid_property
Hybrid properties can also define setter methods. If we wanted
``length`` above, when set, to modify the endpoint value::
- class Interval(object):
+ class Interval:
# ...
@hybrid_property
this, using the :meth:`.hybrid_property.update_expression` decorator.
A handler that works similarly to our setter would be::
- class Interval(object):
+ class Interval:
# ...
@hybrid_property
of the hybrid must match that of how it is accessed. Something
like this wouldn't work::
- class Interval(object):
+ class Interval:
# ...
def _get(self):
from sqlalchemy.ext.hybrid import hybrid_method
- class SomeClass(object):
+ class SomeClass:
@hybrid_method
def value(self, x, y):
return self._value + x + y
from sqlalchemy.ext.hybrid import hybrid_property
- class SomeClass(object):
+ class SomeClass:
@hybrid_property
def value(self):
return self._value
to be used without conflicting with the same-named attributes
normally present on the :class:`.QueryableAttribute`::
- class SuperClass(object):
+ class SuperClass:
# ...
@hybrid_property
orm_instrumentation.instrumentation_finders = instrumentation_finders
-class InstrumentationManager(object):
+class InstrumentationManager:
"""User-defined class instrumentation extension.
:class:`.InstrumentationManager` can be subclassed in order
from sqlalchemy import mapper
- class MyDataClass(object):
+ class MyDataClass:
pass
# associates mutation listeners with MyDataClass.data
Column('y2', Integer),
)
- class Vertex(object):
+ class Vertex:
pass
mapper(Vertex, vertices, properties={
from ..util import memoized_property
-class MutableBase(object):
+class MutableBase:
"""Common base class to :class:`.Mutable`
and :class:`.MutableComposite`.
return cls
-class Identified(object):
+class Identified:
logging_name = None
def _should_log_debug(self):
return self.logger.isEnabledFor(logging.INFO)
-class InstanceLogger(object):
+class InstanceLogger:
"""A logger adapter (wrapper) for :class:`.Identified` subclasses.
This allows multiple instances (e.g. Engine or Pool instances)
instance.logger = logger
-class echo_property(object):
+class echo_property:
__doc__ = """\
When ``True``, enable log output for this element.
OP_MODIFIED = util.symbol("MODIFIED")
-class AttributeEvent(object):
+class AttributeEvent:
"""A token propagated throughout the course of a chain of attribute
events.
Event = AttributeEvent
-class AttributeImpl(object):
+class AttributeImpl:
"""internal implementation for instrumented attributes."""
def __init__(
return mapper
-class InspectionAttr(object):
+class InspectionAttr:
"""A base class applied to all ORM objects that can be returned
by the :func:`_sa.inspect` function.
return {}
-class _MappedAttribute(object):
+class _MappedAttribute:
"""Mixin for attributes which should be replaced by mapper-assigned
attributes.
return not test(thing)
-class _MultipleClassMarker(object):
+class _MultipleClassMarker:
"""refers to multiple classes of the same name
within _decl_class_registry.
self.contents.add(weakref.ref(item, self._remove_item))
-class _ModuleMarker(object):
+class _ModuleMarker:
"""Refers to a module name within
_decl_class_registry.
existing.remove_item(cls)
-class _ModNS(object):
+class _ModNS:
__slots__ = ("__parent",)
def __init__(self, parent):
)
-class _GetColumns(object):
+class _GetColumns:
__slots__ = ("cls",)
def __init__(self, cls):
)
-class _GetTable(object):
+class _GetTable:
__slots__ = "key", "metadata"
def __init__(self, key, metadata):
return _GetColumns(value)
-class _class_resolver(object):
+class _class_resolver:
__slots__ = (
"cls",
"prop",
and return values to events::
from sqlalchemy.orm.collections import collection
- class MyClass(object):
+ class MyClass:
# ...
@collection.adds(1)
__instrumentation_mutex = util.threading.Lock()
-class _PlainColumnGetter(object):
+class _PlainColumnGetter:
"""Plain column getter, stores collection of Column objects
directly.
return key[0]
-class _SerializableColumnGetter(object):
+class _SerializableColumnGetter:
"""Column-based getter used in version 0.7.6 only.
Remains here for pickle compatibility with 0.7.6.
return lambda: MappedCollection(keyfunc)
-class _SerializableAttrGetter(object):
+class _SerializableAttrGetter:
def __init__(self, name):
self.name = name
self.getter = operator.attrgetter(name)
return lambda: MappedCollection(keyfunc)
-class collection(object):
+class collection:
"""Decorators for entity collection classes.
The decorators fall into two groups: annotations and interception recipes.
"""Fetch the :class:`.CollectionAdapter` for a collection."""
-class CollectionAdapter(object):
+class CollectionAdapter:
"""Bridges between the ORM and arbitrary Python collections.
Proxies base-level collection operations (append, remove, iterate)
LABEL_STYLE_LEGACY_ORM = util.symbol("LABEL_STYLE_LEGACY_ORM")
-class QueryContext(object):
+class QueryContext:
__slots__ = (
"compile_state",
"query",
return None
-class _QueryEntity(object):
+class _QueryEntity:
"""represent an entity column returned within a Query result."""
__slots__ = ()
easily reused across different mappings. The example below illustrates
both::
- class ProvidesUser(object):
+ class ProvidesUser:
"A mixin that adds a 'user' relationship to classes."
@declared_attr
Below, both MyClass as well as MySubClass will have a distinct
``id`` Column object established::
- class HasIdMixin(object):
+ class HasIdMixin:
@declared_attr.cascading
def id(cls):
if has_inherited_table(cls):
)
-class registry(object):
+class registry:
"""Generalized registry for mapping classes.
The :class:`_orm.registry` serves as the basis for maintaining a collection
mapper_registry = registry()
@mapper_registry.as_declarative_base()
- class Base(object):
+ class Base:
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
from sqlalchemy.orm import as_declarative
@as_declarative()
- class Base(object):
+ class Base:
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
return False
-class _MapperConfig(object):
+class _MapperConfig:
__slots__ = ("cls", "classname", "properties", "declared_attr_reg")
@classmethod
from .. import util
-class DependencyProcessor(object):
+class DependencyProcessor:
def __init__(self, prop):
self.prop = prop
self.cascade = prop.cascade
def instrument_class(self, mapper):
prop = self
- class _ProxyImpl(object):
+ class _ProxyImpl:
accepts_scalar_loader = False
load_on_unexpire = True
collection = False
% (self.parent, self.key, self.parent)
)
- class NoninheritedConcreteProp(object):
+ class NoninheritedConcreteProp:
def __set__(s, obj, value):
warn()
Column('job_status', String(50))
)
- class MyClass(object):
+ class MyClass:
@property
def _job_status_descriptor(self):
return "Status: %s" % self._job_status
self.remove(state, dict_, value, initiator, passive=passive)
-class DynamicCollectionAdapter(object):
+class DynamicCollectionAdapter:
"""simplified CollectionAdapter for internal API consistency"""
def __init__(self, data):
__nonzero__ = __bool__
-class AppenderMixin(object):
+class AppenderMixin:
query_class = None
def __init__(self, attr, state):
return type(name, (AppenderMixin, cls), {"query_class": cls})
-class CollectionHistory(object):
+class CollectionHistory:
"""Overrides AttributeHistory to receive append/remove events directly."""
def __init__(self, attr, state, apply_to=None):
)
-class EvaluatorCompiler(object):
+class EvaluatorCompiler:
def __init__(self, target_cls=None):
self.target_cls = target_cls
"""Called when an attribute is instrumented."""
-class _InstrumentationEventsHold(object):
+class _InstrumentationEventsHold:
"""temporary marker object used to transfer from _accept_with() to
_listen() on the InstrumentationEvents class.
def _clear(cls):
cls.all_holds.clear()
- class HoldEvents(object):
+ class HoldEvents:
_dispatch_target = None
@classmethod
from .. import exc as sa_exc
-class IdentityMap(object):
+class IdentityMap:
def __init__(self):
self._dict = {}
self._modified = set()
)
-class _SerializeManager(object):
+class _SerializeManager:
"""Provide serialization of a :class:`.ClassManager`.
The :class:`.InstanceState` uses ``__init__()`` on serialize
manager.dispatch.unpickle(state, state_dict)
-class InstrumentationFactory(object):
+class InstrumentationFactory:
"""Factory for new ClassManager instances."""
def create_manager_for_cls(self, class_):
self.process_query(query)
-class LoaderStrategy(object):
+class LoaderStrategy:
"""Describe the loading behavior of a StrategizedProperty object.
The ``LoaderStrategy`` interacts with the querying process in three
return polymorphic_instance
-class PostLoad(object):
+class PostLoad:
"""Track loaders and states for "post load" operations."""
__slots__ = "loaders", "states", "load_keys"
pass
-class BulkUD(object):
+class BulkUD:
"""State used for the orm.Query version of update() / delete().
This object is now specific to Query only.
return element
-class JoinCondition(object):
+class JoinCondition:
def __init__(
self,
parent_persist_selectable,
return lazywhere, bind_to_col, equated_columns
-class _ColInAnnotations(object):
+class _ColInAnnotations:
"""Serializable object that tests for a name in c._annotations."""
__slots__ = ("name",)
__all__ = ["scoped_session", "ScopedSessionMixin"]
-class ScopedSessionMixin(object):
+class ScopedSessionMixin:
@property
def _proxied(self):
return self.registry()
Session = scoped_session(sessionmaker())
- class MyClass(object):
+ class MyClass:
query = Session.query_property()
# after mappers are defined
"""
- class query(object):
+ class query:
def __get__(s, instance, owner):
try:
mapper = class_mapper(owner)
return state.session
-class _SessionClassMethods(object):
+class _SessionClassMethods:
"""Class-level methods for :class:`.Session`, :class:`.sessionmaker`."""
@classmethod
state._strong_obj = None
-class AttributeState(object):
+class AttributeState:
"""Provide an inspection interface corresponding
to a particular attribute on a particular mapped object.
return self.state.get_history(self.key, PASSIVE_OFF ^ INIT_OK)
-class PendingCollection(object):
+class PendingCollection:
"""A writable placeholder for an unloaded collection.
Stores items appended to and removed from a collection that has not yet
)
-class LoadDeferredColumns(object):
+class LoadDeferredColumns:
"""serializable loader object used by DeferredColumnLoader"""
def __init__(self, key, raiseload=False):
populators["new"].append((self.key, reset_for_lazy_callable))
-class LoadLazyAttribute(object):
+class LoadLazyAttribute:
"""semi-serializable loader object used by LazyLoader
Historically, this object would be carried along with instances that
return q
- class _SubqCollections(object):
+ class _SubqCollections:
"""Given a :class:`_query.Query` used to emit the "subquery load",
provide a load interface that executes the query at the
first moment a value is needed.
return None
-class loader_option(object):
+class loader_option:
def __init__(self):
pass
event.listen(descriptor, "set", set_, raw=True, retval=True)
-class UOWTransaction(object):
+class UOWTransaction:
def __init__(self, session):
self.session = session
self.session._register_persistent(other)
-class IterateMappersMixin(object):
+class IterateMappersMixin:
def _mappers(self, uow):
if self.fromparent:
return iter(
return False
-class PostSortRec(object):
+class PostSortRec:
__slots__ = ("disabled",)
def __new__(cls, uow, *args):
return not entity or entity.isa(self.mapper)
-class AliasedClass(object):
+class AliasedClass:
r"""Represents an "aliased" form of a mapped class for usage with Query.
The ORM equivalent of a :func:`~sqlalchemy.sql.expression.alias`
return "aliased(%s)" % (self._target.__name__,)
-class _WrapUserEntity(object):
+class _WrapUserEntity:
"""A wrapper used within the loader_criteria lambda caller so that
we can bypass declared_attr descriptors on unmapped mixins, which
normally emit a warning for such use.
reset_none = util.symbol("reset_none")
-class _ConnDialect(object):
+class _ConnDialect:
"""partial implementation of :class:`.Dialect`
which provides DBAPI connection methods.
raise NotImplementedError()
-class _ConnectionRecord(object):
+class _ConnectionRecord:
"""Internal object which maintains an individual DBAPI connection
referenced by a :class:`_pool.Pool`.
_strong_ref_connection_records = {}
-class _ConnectionFairy(object):
+class _ConnectionFairy:
"""Proxies a DBAPI connection and provides return-on-dereference
support.
proxies.clear()
-class _DBProxy(object):
+class _DBProxy:
"""Layers connection pooling behavior on top of a standard DB-API module.
EMPTY_ANNOTATIONS = util.immutabledict()
-class SupportsAnnotations(object):
+class SupportsAnnotations:
_annotations = EMPTY_ANNOTATIONS
@util.memoized_property
return self
-class Annotated(object):
+class Annotated:
"""clones a SupportsAnnotated and applies an 'annotations' dictionary.
Unlike regular clones, this clone also mimics __hash__() and
NO_ARG = util.symbol("NO_ARG")
-class Immutable(object):
+class Immutable:
"""mark a ClauseElement as 'immutable' when expressions are cloned."""
_is_immutable = True
return dict(dialect_cls.construct_arguments)
-class DialectKWArgs(object):
+class DialectKWArgs:
"""Establish the ability for a class to have dialect-specific arguments
with defaults and constructor validation.
construct_arg_dictionary[arg_name] = kwargs[k]
-class CompileState(object):
+class CompileState:
"""Produces additional object state necessary for a statement to be
compiled.
return value
-class SchemaEventTarget(object):
+class SchemaEventTarget:
"""Base class for elements that are the targets of :class:`.DDLEvents`
events.
__traverse_options__ = {"schema_visitor": True}
-class ColumnCollection(object):
+class ColumnCollection:
"""Collection of :class:`_expression.ColumnElement` instances,
typically for
:class:`_sql.FromClause` objects.
yield resolved, column, strname, add_element
-class RoleImpl(object):
+class RoleImpl:
__slots__ = ("_role_class", "name", "_use_inspection")
def _literal_coercion(self, element, **kw):
util.raise_(exc.ArgumentError(msg, code=code), replace_context=err)
-class _Deannotate(object):
+class _Deannotate:
__slots__ = ()
def _post_coercion(self, resolved, **kw):
return _deep_deannotate(resolved)
-class _StringOnly(object):
+class _StringOnly:
__slots__ = ()
_resolve_literal_only = True
-class _ReturnsStringKey(object):
+class _ReturnsStringKey:
__slots__ = ()
def _implicit_coercions(
return element
-class _ColumnCoercions(object):
+class _ColumnCoercions:
__slots__ = ()
def _warn_for_scalar_subquery_coercion(self):
)
-class _NoTextCoercion(object):
+class _NoTextCoercion:
__slots__ = ()
def _literal_coercion(self, element, argname=None, **kw):
self._raise_for_expected(element, argname)
-class _CoerceLiterals(object):
+class _CoerceLiterals:
__slots__ = ()
_coerce_consts = False
_coerce_star = False
return element
-class _SelectIsNotFrom(object):
+class _SelectIsNotFrom:
__slots__ = ()
def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
util.warn(message)
-class Compiled(object):
+class Compiled:
"""Represent a compiled SQL or DDL expression.
return get_col_spec(**kw)
-class IdentifierPreparer(object):
+class IdentifierPreparer:
"""Handle quoting and case-folding of identifiers based on options."""
self.select = coercions.expect(roles.DMLSelectRole, select)
-class DMLWhereBase(object):
+class DMLWhereBase:
_where_criteria = ()
@_generative
return self._anon_label(label, add_hash=idx)
-class WrapsColumnExpression(object):
+class WrapsColumnExpression:
"""Mixin that defines a :class:`_expression.ColumnElement`
as a wrapper with special
labeling behavior for an expression that already has a name.
self.type = sqltypes.to_instance(type_)
-class _FunctionGenerator(object):
+class _FunctionGenerator:
"""Generate SQL function expressions.
:data:`.func` is a special object instance which generates SQL
return fn(self.parent_lambda._resolved)
-class AnalyzedCode(object):
+class AnalyzedCode:
__slots__ = (
"track_closure_variables",
"track_bound_values",
)
-class NonAnalyzedFunction(object):
+class NonAnalyzedFunction:
__slots__ = ("expr",)
closure_bindparams = None
return self.expr
-class AnalyzedFunction(object):
+class AnalyzedFunction:
__slots__ = (
"analyzed_code",
"fn",
from .. import exc
-class ConventionDict(object):
+class ConventionDict:
def __init__(self, const, table, convention):
self.const = const
self._is_fk = isinstance(const, ForeignKeyConstraint)
from .. import util
-class Operators(object):
+class Operators:
"""Base of comparison and logical operators.
Implements base methods
raise NotImplementedError(str(op))
-class custom_op(object):
+class custom_op:
"""Represent a 'custom' operator.
:class:`.custom_op` is normally instantiated when the
from .. import util
-class SQLRole(object):
+class SQLRole:
"""Define a "role" within a SQL statement structure.
Classes within SQL Core participate within SQLRole hierarchies in order
uses_inspection = False
-class UsesInspection(object):
+class UsesInspection:
_post_inspect = None
uses_inspection = True
-class AllowsLambdaRole(object):
+class AllowsLambdaRole:
allows_lambda = True
return "ColumnDefault(%r)" % (self.arg,)
-class IdentityOptions(object):
+class IdentityOptions:
"""Defines options for a named database sequence or an identity column.
.. versionadded:: 1.3.18
raise NotImplementedError()
-class ColumnCollectionMixin(object):
+class ColumnCollectionMixin:
columns = None
"""A :class:`_expression.ColumnCollection` of :class:`_schema.Column`
)
-class HasPrefixes(object):
+class HasPrefixes:
_prefixes = ()
_has_prefixes_traverse_internals = [
)
-class HasSuffixes(object):
+class HasSuffixes:
_suffixes = ()
_has_suffixes_traverse_internals = [
)
-class HasHints(object):
+class HasHints:
_hints = util.immutabledict()
_statement_hints = ()
return [self] + self.left._from_objects + self.right._from_objects
-class NoInit(object):
+class NoInit:
def __init__(self, *arg, **kw):
raise NotImplementedError(
"The %s class is not intended to be constructed "
return self.element._from_objects
-class DeprecatedSelectBaseGenerations(object):
+class DeprecatedSelectBaseGenerations:
"""A collection of methods available on :class:`_sql.Select` and
:class:`_sql.CompoundSelect`, these are all **deprecated** methods as they
modify the object in-place.
self._bind = bind
-class DeprecatedSelectGenerations(object):
+class DeprecatedSelectGenerations:
"""A collection of methods available on :class:`_sql.Select`, these
are all **deprecated** methods as they modify the :class:`_sql.Select`
object in -place.
return replace_from_obj_index
-class _SelectFromElements(object):
+class _SelectFromElements:
def _iterate_from_elements(self):
# note this does not include elements
# in _setup_joins or _legacy_setup_joins
from ..util import pickle
-class _LookupExpressionAdapter(object):
+class _LookupExpressionAdapter:
"""Mixin expression adaptations based on lookup tables.
comparator_factory = Comparator
-class Concatenable(object):
+class Concatenable:
"""A mixin that marks a type as supporting 'concatenation',
typically strings."""
comparator_factory = Comparator
-class Indexable(object):
+class Indexable:
"""A mixin that marks a type as supporting indexing operations,
such as array or JSON structures.
)
-class HasCacheKey(object):
+class HasCacheKey:
_cache_key_traversal = NO_CACHE
__slots__ = ()
_cache_key_traversal_visitor = _CacheKey()
-class HasCopyInternals(object):
+class HasCopyInternals:
def _clone(self, **kw):
raise NotImplementedError()
pass
-class ExternalType(object):
+class ExternalType:
"""mixin that defines attributes and behaviors specific to third-party
datatypes.
return self
-class Emulated(object):
+class Emulated:
"""Mixin for base types that emulate the behavior of a DB-native type.
An :class:`.Emulated` type will use an available database type
return super(Emulated, self).adapt(impltype, **kw)
-class NativeForEmulated(object):
+class NativeForEmulated:
"""Indicates DB-native types supported by an :class:`.Emulated` type.
.. versionadded:: 1.2.0b3
return repr(element)
-class _repr_base(object):
+class _repr_base:
_LIST = 0
_TUPLE = 1
_DICT = 2
self.allow_label_resolve = allow_label_resolve
self._wrap = None
- class _IncludeExcludeMapping(object):
+ class _IncludeExcludeMapping:
def __init__(self, parent, columns):
self.parent = parent
self.columns = columns
HasCacheKey objects."""
-class ExternalTraversal(object):
+class ExternalTraversal:
"""Base class for visitor objects which can traverse externally using
the :func:`.visitors.traverse` function.
return ec.error
-class _ErrorContainer(object):
+class _ErrorContainer:
error = None
return _expect_raises(except_cls, msg=msg, check_context=check_context)
-class AssertsCompiledSQL(object):
+class AssertsCompiledSQL:
def assert_compile(
self,
clause,
if compile_kwargs:
kw["compile_kwargs"] = compile_kwargs
- class DontAccess(object):
+ class DontAccess:
def __getattribute__(self, key):
raise NotImplementedError(
"compiler accessed .statement; use "
"compiler.current_executable"
)
- class CheckCompilerAccess(object):
+ class CheckCompilerAccess:
def __init__(self, test_statement):
self.test_statement = test_statement
self._annotations = {}
)
-class ComparesTables(object):
+class ComparesTables:
def assert_tables_equal(self, table, reflected_table, strict_types=False):
assert len(table.c) == len(reflected_table.c)
for c, reflected_c in zip(table.c, reflected_table.c):
)
-class AssertsExecutionResults(object):
+class AssertsExecutionResults:
def assert_result(self, result, class_, *objects):
result = list(result)
print(repr(result))
from ..schema import _DDLCompiles
-class AssertRule(object):
+class AssertRule:
is_consumed = False
errormessage = None
self.errormessage = list(self.rules)[0].errormessage
-class SQLExecuteObserved(object):
+class SQLExecuteObserved:
def __init__(self, context, clauseelement, multiparams, params):
self.context = context
self.clauseelement = clauseelement
pass
-class SQLAsserter(object):
+class SQLAsserter:
def __init__(self):
self.accumulated = []
return _fixture_functions.mark_base_test_class()
-class Config(object):
+class Config:
def __init__(self, db, db_opts, options, file_config):
self._set_name(db)
self.db = db
from .. import pool
-class ConnectionKiller(object):
+class ConnectionKiller:
def __init__(self):
self.proxy_refs = weakref.WeakKeyDictionary()
self.testing_engines = collections.defaultdict(set)
yield mod.dialect()
-class ReconnectFixture(object):
+class ReconnectFixture:
def __init__(self, dbapi):
self.dbapi = dbapi
self.connections = []
return engine
-class DBAPIProxyCursor(object):
+class DBAPIProxyCursor:
"""Proxy a DBAPI cursor.
Tests can provide subclasses of this to intercept
return getattr(self.cursor, key)
-class DBAPIProxyConnection(object):
+class DBAPIProxyConnection:
"""Proxy a DBAPI connection.
Tests can provide subclasses of this to intercept
_repr_stack = set()
-class BasicEntity(object):
+class BasicEntity:
def __init__(self, **kw):
for key, value in kw.items():
setattr(self, key, value)
_recursion_stack = set()
-class ComparableMixin(object):
+class ComparableMixin:
def __ne__(self, other):
return not self.__eq__(other)
return rule
-class compound(object):
+class compound:
def __init__(self):
self.fails = set()
self.skips = set()
return fails_if(NotPredicate(predicate), reason)
-class Predicate(object):
+class Predicate:
@classmethod
def as_predicate(cls, predicate, description=None):
if isinstance(predicate, compound):
@config.mark_base_test_class()
-class TestBase(object):
+class TestBase:
# A sequence of requirement names matching testing.requires decorators
__requires__ = ()
_connection_fixture_connection = None
-class FutureEngineMixin(object):
+class FutureEngineMixin:
"""alembic's suite still using this"""
)
-class NoCache(object):
+class NoCache:
@config.fixture(autouse=True, scope="function")
def _disable_cache(self):
_cache = config.db._compiled_cache
config.db._compiled_cache = _cache
-class RemovesEvents(object):
+class RemovesEvents:
@util.memoized_property
def _event_fns(self):
return set()
cls_registry[classname] = cls
DeclarativeMeta.__init__(cls, classname, bases, dict_)
- class DeclarativeBasic(object):
+ class DeclarativeBasic:
__table_cls__ = schema.Table
_DeclBase = declarative_base(
pass
-class Screen(object):
+class Screen:
def __init__(self, obj, parent=None):
self.obj = obj
self.parent = parent
-class Foo(object):
+class Foo:
def __init__(self, moredata, stuff="im stuff"):
self.data = "im data"
self.stuff = stuff
)
-class Bar(object):
+class Bar:
def __init__(self, x, y):
self.x = x
self.y = y
self.y = y
-class BarWithoutCompare(object):
+class BarWithoutCompare:
def __init__(self, x, y):
self.x = x
self.y = y
return "Bar(%d, %d)" % (self.x, self.y)
-class NotComparable(object):
+class NotComparable:
def __init__(self, data):
self.data = data
return NotImplemented
-class BrokenComparable(object):
+class BrokenComparable:
def __init__(self, data):
self.data = data
import ConfigParser as configparser
import collections as collections_abc # noqa
- class ABC(object):
+ class ABC:
__metaclass__ = abc.ABCMeta
_profile_stats.reset_count()
-class ProfileStatsFile(object):
+class ProfileStatsFile:
"""Store per-platform/fn profiling results in a file.
There was no json module available when this was written, but now
FOLLOWER_IDENT = None
-class register(object):
+class register:
def __init__(self):
self.fns = {}
from ..pool import QueuePool
-class Requirements(object):
+class Requirements:
pass
return col
-class eq_type_affinity(object):
+class eq_type_affinity:
"""Helper to compare types inside of datastructures based on affinity.
E.g.::
return self.target._type_affinity is not other._type_affinity
-class eq_clause_element(object):
+class eq_clause_element:
"""Helper to compare SQL structures based on compare()"""
def __init__(self, target):
from ...util import u
-class _LiteralRoundTripFixture(object):
+class _LiteralRoundTripFixture:
supports_whereclause = True
@testing.fixture
EMPTY_SET = frozenset()
-class ImmutableContainer(object):
+class ImmutableContainer:
def _immutable(self, *arg, **kw):
raise TypeError("%s object is immutable" % self.__class__.__name__)
return "FacadeDict(%s)" % dict.__repr__(self)
-class Properties(object):
+class Properties:
"""Provide a __getattr__/__setattr__ interface over a dict."""
__slots__ = ("_data",)
__isub__ = difference_update
-class IdentitySet(object):
+class IdentitySet:
"""A set that considers only object id() for uniqueness.
This strategy has edge cases for builtin types- it's possible to have
return "%s(%r)" % (type(self).__name__, list(self._members.values()))
-class WeakSequence(object):
+class WeakSequence:
def __init__(self, __elements=()):
# adapted from weakref.WeakKeyDictionary, prevent reference
# cycles in the collection itself
]
-class UniqueAppender(object):
+class UniqueAppender:
"""Appends items to a collection ensuring uniqueness.
Additional appends() of the same object are ignored. Membership is
self._mutex.release()
-class ScopedRegistry(object):
+class ScopedRegistry:
"""A Registry that can store one or multiple instances of a single
class on the basis of a "scope" function.
)
-class nullcontext(object):
+class nullcontext:
"""Context manager that does no additional processing.
Vendored from Python 3.7.
from abc import ABCMeta
- class ABC(object):
+ class ABC:
__metaclass__ = ABCMeta
try:
return m.hexdigest()
-class safe_reraise(object):
+class safe_reraise:
"""Reraise an exception after invoking some
handler code.
return decorated
-class PluginLoader(object):
+class PluginLoader:
def __init__(self, group, auto_fn=None):
self.group = group
self.impls = {}
return "%s(%s)" % (obj.__class__.__name__, ", ".join(output))
-class portable_instancemethod(object):
+class portable_instancemethod:
"""Turn an instancemethod into a (parent, name) pair
to produce a serializable callable.
% (obj, qualifier, ", ".join(interface))
)
- class AnonymousInterface(object):
+ class AnonymousInterface:
"""A callable-holding shell."""
if cls:
)
-class memoized_property(object):
+class memoized_property:
"""A read-only @property that is only evaluated once."""
def __init__(self, fget, doc=None):
return update_wrapper(oneshot, fn)
-class HasMemoized(object):
+class HasMemoized:
"""A class that maintains the names of memoized elements in a
collection for easy cache clearing, generative, etc.
self.__dict__[key] = value
self._memoized_keys |= {key}
- class memoized_attribute(object):
+ class memoized_attribute:
"""A read-only @property that is only evaluated once."""
def __init__(self, fget, doc=None):
return update_wrapper(oneshot, fn)
-class MemoizedSlots(object):
+class MemoizedSlots:
"""Apply memoized items to an object using a __getattr__ scheme.
This allows the functionality of memoized_property and
return desc.fget(cls)
-class hybridproperty(object):
+class hybridproperty:
def __init__(self, func):
self.func = func
self.clslevel = func
return self
-class hybridmethod(object):
+class hybridmethod:
"""Decorate a function as cls- or instance- level."""
def __init__(self, func):
_symbol.__name__ = "symbol"
-class symbol(object):
+class symbol:
"""A constant symbol.
>>> symbol('foo') is symbol('foo')
@testing.fails()
def test_fixture_failure(self):
- class Foo(object):
+ class Foo:
pass
stuff = []
# ensure a pure growing test trips the assertion
@testing.fails_if(lambda: True)
def test_fixture(self):
- class Foo(object):
+ class Foo:
pass
x = []
*[Column("col%d" % i, Integer) for i in range(10)]
)
- class Wide(object):
+ class Wide:
pass
self.mapper_registry.map_imperatively(
),
)
- class SomeClass(object):
+ class SomeClass:
pass
self.mapper_registry.map_imperatively(SomeClass, some_table)
Column("t1id", ForeignKey("t1.id")),
)
- class T1(object):
+ class T1:
pass
t1_mapper = self.mapper_registry.map_imperatively(T1, t1)
@testing.emits_warning()
@profile_memory()
def go():
- class T2(object):
+ class T2:
pass
t2_mapper = self.mapper_registry.map_imperatively(T2, t2)
Column("t1id", Integer, ForeignKey("table1.id")),
)
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
self.mapper_registry.map_imperatively(
Column("t1id", Integer, ForeignKey("table1.id")),
)
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
self.mapper_registry.map_imperatively(
users = self.tables.users
- class Foo(object):
+ class Foo:
@hybrid.hybrid_property
def user_name(self):
return self.name
go()
def test_weak_sequence(self):
- class Foo(object):
+ class Foo:
pass
f = Foo()
__requires__ = ("cpython", "python_profiling_backend")
def setup_test(self):
- class SomeEnum(object):
+ class SomeEnum:
# Implements PEP 435 in the minimal fashion needed by SQLAlchemy
_members = {}
class QueuePoolTest(fixtures.TestBase, AssertsExecutionResults):
__requires__ = ("cpython", "python_profiling_backend")
- class Connection(object):
+ class Connection:
def rollback(self):
pass
__backend__ = True
def _rowproxy_fixture(self, keys, processors, row, row_cls):
- class MockMeta(object):
+ class MockMeta:
def __init__(self):
pass
self._test_getitem_value_refcounts_new(tuple)
def test_value_refcounts_custom_seq(self):
- class CustomSeq(object):
+ class CustomSeq:
def __init__(self, data):
self.data = data
from sqlalchemy.testing.util import gc_collect
-class TearDownLocalEventsFixture(object):
+class TearDownLocalEventsFixture:
def teardown_test(self):
classes = set()
for entry in event.base._registrars.values():
def event_three(self, x):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
self.Target = Target
class SlotsEventsTest(fixtures.TestBase):
def test_no_slots_dispatch(self):
- class Target(object):
+ class Target:
__slots__ = ()
class TargetEvents(event.Events):
event.listen(t1, "event_one", Mock())
def test_slots_dispatch(self):
- class Target(object):
+ class Target:
__slots__ = ("_slots_dispatch",)
class TargetEvents(event.Events):
def event_five(self, x, y, z, q):
pass
- class TargetOne(object):
+ class TargetOne:
dispatch = event.dispatcher(TargetEventsOne)
return TargetOne
def event_five(self, x, y, z, q):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
return Target
def event_six(self, x, y):
pass
- class TargetOne(object):
+ class TargetOne:
dispatch = event.dispatcher(TargetEventsOne)
self.TargetOne = TargetOne
def test_legacy_accept_from_method(self):
canary = Mock()
- class MyClass(object):
+ class MyClass:
def handler1(self, x, y):
canary(x, y)
def event_one(self, x, y):
pass
- class TargetOne(object):
+ class TargetOne:
dispatch = event.dispatcher(TargetEventsOne)
self.TargetOne = TargetOne
def event_one(self, x, y):
pass
- class TargetOne(object):
+ class TargetOne:
dispatch = event.dispatcher(TargetEventsOne)
- class TargetTwo(object):
+ class TargetTwo:
dispatch = event.dispatcher(TargetEventsTwo)
self.TargetOne = TargetOne
def event_one(self, x, y):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
self.Target = Target
def some_event(self, x, y):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
self.Target = Target
def event_one(self, x, y):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
self.Target = Target
def event_two(self, arg):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
self.Target = Target
def event_one(self, target, arg):
pass
- class BaseTarget(object):
+ class BaseTarget:
dispatch = event.dispatcher(TargetEvents)
class TargetFactory(BaseTarget):
def event_one(self, target, arg):
pass
- class BaseTarget(object):
+ class BaseTarget:
dispatch = event.dispatcher(TargetEvents)
class SubTarget(BaseTarget):
def event_three(self, x):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
return Target
def event_one(self, x):
pass
- class Target(object):
+ class Target:
dispatch = event.dispatcher(TargetEvents)
return Target
def test_instance(self):
Target = self._fixture()
- class Foo(object):
+ class Foo:
def __init__(self):
self.mock = Mock()
from sqlalchemy.testing import fixtures
-class TestFixture(object):
+class TestFixture:
pass
class SomeFoo(TestFixture):
pass
- class SomeFooInspect(object):
+ class SomeFooInspect:
def __init__(self, target):
self.target = target
def _setup_logger(self):
rootlogger = logging.getLogger("sqlalchemy.engine.Engine")
- class MyStream(object):
+ class MyStream:
def write(self, string):
sys.stdout.write(string)
sys.stdout.flush()
class WeakSequenceTest(fixtures.TestBase):
@testing.requires.predictable_gc
def test_cleanout_elements(self):
- class Foo(object):
+ class Foo:
pass
f1, f2, f3 = Foo(), Foo(), Foo()
@testing.requires.predictable_gc
def test_cleanout_appended(self):
- class Foo(object):
+ class Foo:
pass
f1, f2, f3 = Foo(), Foo(), Foo()
def test_memoized_property(self):
val = [20]
- class Foo(object):
+ class Foo:
@util.memoized_property
def bar(self):
v = val[0]
def test_memoized_instancemethod(self):
val = [20]
- class Foo(object):
+ class Foo:
@util.memoized_instancemethod
def bar(self):
v = val[0]
eq_(c.__doc__, None)
def test_wrapping_update_wrapper_cls(self):
- class MyFancyDefault(object):
+ class MyFancyDefault:
"""a fancy default"""
def __call__(self):
eq_(c.__doc__, "run the fancy default")
def test_wrapping_update_wrapper_cls_noclsdocstring(self):
- class MyFancyDefault(object):
+ class MyFancyDefault:
def __call__(self):
"""run the fancy default"""
return 10
eq_(c.__doc__, "run the fancy default")
def test_wrapping_update_wrapper_cls_nomethdocstring(self):
- class MyFancyDefault(object):
+ class MyFancyDefault:
"""a fancy default"""
def __call__(self):
eq_(c.__doc__, "a fancy default")
def test_wrapping_update_wrapper_cls_noclsdocstring_nomethdocstring(self):
- class MyFancyDefault(object):
+ class MyFancyDefault:
def __call__(self):
return 10
class LRUTest(fixtures.TestBase):
def test_lru(self):
- class item(object):
+ class item:
def __init__(self, id_):
self.id = id_
assert list(util.flatten_iterator(iter_list)) == ["asdf", "x", "y"]
-class HashOverride(object):
+class HashOverride:
def __init__(self, value=None):
self.value = value
return hash(self.value)
-class NoHash(object):
+class NoHash:
def __init__(self, value=None):
self.value = value
__hash__ = None
-class EqOverride(object):
+class EqOverride:
def __init__(self, value=None):
self.value = value
return True
-class HashEqOverride(object):
+class HashEqOverride:
def __init__(self, value=None):
self.value = value
self._notok(object())
def test_duck_2(self):
- class duck2(object):
+ class duck2:
def items(duck):
return list(self.baseline)
self._ok(duck2())
def test_duck_4(self):
- class duck4(object):
+ class duck4:
def iterkeys(duck):
return iter(["a", "b", "c"])
self._notok(duck4())
def test_duck_5(self):
- class duck5(object):
+ class duck5:
def keys(duck):
return ["a", "b", "c"]
self._ok(duck5())
def test_duck_6(self):
- class duck6(object):
+ class duck6:
def keys(duck):
return ["a", "b", "c"]
class DuckTypeCollectionTest(fixtures.TestBase):
def test_sets(self):
- class SetLike(object):
+ class SetLike:
def add(self):
pass
class PublicFactoryTest(fixtures.TestBase):
def _fixture(self):
- class Thingy(object):
+ class Thingy:
def __init__(self, value):
"make a thingy"
self.value = value
class ArgInspectionTest(fixtures.TestBase):
def test_get_cls_kwargs(self):
- class A(object):
+ class A:
def __init__(self, a):
pass
def __init__(self, a11, **kw):
pass
- class B(object):
+ class B:
def __init__(self, b, **kw):
pass
class CB2A(B2, A):
pass
- class D(object):
+ class D:
pass
class BA2(B, A):
assert_raises(TypeError, get_callable_argspec, object)
def test_callable_argspec_method(self):
- class Foo(object):
+ class Foo:
def foo(self, x, y, **kw):
pass
)
def test_callable_argspec_instance_method_no_self(self):
- class Foo(object):
+ class Foo:
def foo(self, x, y, **kw):
pass
)
def test_callable_argspec_unbound_method_no_self(self):
- class Foo(object):
+ class Foo:
def foo(self, x, y, **kw):
pass
)
def test_callable_argspec_init(self):
- class Foo(object):
+ class Foo:
def __init__(self, x, y):
pass
)
def test_callable_argspec_init_no_self(self):
- class Foo(object):
+ class Foo:
def __init__(self, x, y):
pass
)
def test_callable_argspec_call(self):
- class Foo(object):
+ class Foo:
def __call__(self, x, y):
pass
)
def test_callable_argspec_call_no_self(self):
- class Foo(object):
+ class Foo:
def __call__(self, x, y):
pass
)
-class _Py3KFixtures(object):
+class _Py3KFixtures:
def _kw_only_fixture(self):
pass
parsed = util.format_argspec_init(fn, grouped=grouped)
eq_(parsed, wanted)
- class Obj(object):
+ class Obj:
pass
test(Obj.__init__, object_spec)
- class Obj(object):
+ class Obj:
def __init__(self):
pass
test(Obj.__init__, object_spec)
- class Obj(object):
+ class Obj:
def __init__(slef, a=123):
pass
class GenericReprTest(fixtures.TestBase):
def test_all_positional(self):
- class Foo(object):
+ class Foo:
def __init__(self, a, b, c):
self.a = a
self.b = b
eq_(util.generic_repr(Foo(1, 2, 3)), "Foo(1, 2, 3)")
def test_positional_plus_kw(self):
- class Foo(object):
+ class Foo:
def __init__(self, a, b, c=5, d=4):
self.a = a
self.b = b
eq_(util.generic_repr(Foo(1, 2, 3, 6)), "Foo(1, 2, c=3, d=6)")
def test_kw_defaults(self):
- class Foo(object):
+ class Foo:
def __init__(self, a=1, b=2, c=3, d=4):
self.a = a
self.b = b
eq_(util.generic_repr(Foo(1, 5, 3, 7)), "Foo(b=5, d=7)")
def test_multi_kw(self):
- class Foo(object):
+ class Foo:
def __init__(self, a, b, c=3, d=4):
self.a = a
self.b = b
)
def test_multi_kw_repeated(self):
- class Foo(object):
+ class Foo:
def __init__(self, a=1, b=2):
self.a = a
self.b = b
)
def test_discard_vargs(self):
- class Foo(object):
+ class Foo:
def __init__(self, a, b, *args):
self.a = a
self.b = b
eq_(util.generic_repr(Foo(1, 2, 3, 4)), "Foo(1, 2)")
def test_discard_vargs_kwargs(self):
- class Foo(object):
+ class Foo:
def __init__(self, a, b, *args, **kw):
self.a = a
self.b = b
eq_(util.generic_repr(Foo(1, 2, 3, 4, x=7, y=4)), "Foo(1, 2)")
def test_significant_vargs(self):
- class Foo(object):
+ class Foo:
def __init__(self, a, b, *args):
self.a = a
self.b = b
eq_(util.generic_repr(Foo(1, 2, 3, 4)), "Foo(1, 2, 3, 4)")
def test_no_args(self):
- class Foo(object):
+ class Foo:
def __init__(self):
pass
eq_(util.generic_repr(Foo()), "Foo()")
def test_no_init(self):
- class Foo(object):
+ class Foo:
pass
eq_(util.generic_repr(Foo()), "Foo()")
class AsInterfaceTest(fixtures.TestBase):
- class Something(object):
+ class Something:
def _ignoreme(self):
pass
def bar(self):
pass
- class Partial(object):
+ class Partial:
def bar(self):
pass
- class Object(object):
+ class Object:
pass
def test_no_cls_no_methods(self):
eq_(set(util.class_hierarchy(object)), set((object,)))
def test_single(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
eq_(set(util.class_hierarchy(A)), set((A, object)))
class TestClassProperty(fixtures.TestBase):
def test_simple(self):
- class A(object):
+ class A:
something = {"foo": 1}
class B(A):
class MethodOveriddenTest(fixtures.TestBase):
def test_subclass_overrides_cls_given(self):
- class Foo(object):
+ class Foo:
def bar(self):
pass
is_true(util.method_is_overridden(Bar, Foo.bar))
def test_subclass_overrides(self):
- class Foo(object):
+ class Foo:
def bar(self):
pass
is_true(util.method_is_overridden(Bar(), Foo.bar))
def test_subclass_overrides_skiplevel(self):
- class Foo(object):
+ class Foo:
def bar(self):
pass
is_true(util.method_is_overridden(Bat(), Foo.bar))
def test_subclass_overrides_twolevels(self):
- class Foo(object):
+ class Foo:
def bar(self):
pass
is_true(util.method_is_overridden(Bat(), Foo.bar))
def test_subclass_doesnt_override_cls_given(self):
- class Foo(object):
+ class Foo:
def bar(self):
pass
is_false(util.method_is_overridden(Bar, Foo.bar))
def test_subclass_doesnt_override(self):
- class Foo(object):
+ class Foo:
def bar(self):
pass
is_false(util.method_is_overridden(Bar(), Foo.bar))
def test_subclass_overrides_multi_mro(self):
- class Base(object):
+ class Base:
pass
- class Foo(object):
+ class Foo:
pass
class Bat(Base):
eq_(connection.scalar(table.select()), 7)
-class Foo(object):
+class Foo:
def __init__(self, **kw):
for k in kw:
setattr(self, k, kw[k])
__dialect__ = mysql.dialect()
__backend__ = True
- class SomeEnum(object):
+ class SomeEnum:
# Implements PEP 435 in the minimal fashion needed by SQLAlchemy
__members__ = OrderedDict()
from setting up cx_oracle.CLOBs on
string-based bind params [ticket:793]."""
- class FakeDBAPI(object):
+ class FakeDBAPI:
def __getattr__(self, attr):
return attr
)
t3 = Table("t3", m, Column("foo", TestTypeDec()))
- class CursorWrapper(object):
+ class CursorWrapper:
# cx_oracle cursor can't be modified so we have to
# invent a whole wrapping scheme
def test_query_on_columns_subquery(self):
sess = Session()
- class Foo(object):
+ class Foo:
pass
clear_mappers()
)
def test_query_distinct_on_aliased(self):
- class Foo(object):
+ class Foo:
pass
self.mapper_registry.map_imperatively(Foo, self.table)
eq_(errmsg.orig.sqlstate, "23505")
-class ExecuteManyMode(object):
+class ExecuteManyMode:
__only_on__ = "postgresql+psycopg2"
__backend__ = True
def test_on_conflict_do_update_clauseelem_as_key_set(self, connection):
users = self.tables.users
- class MyElem(object):
+ class MyElem:
def __init__(self, expr):
self.expr = expr
from sqlalchemy.testing.assertions import is_true
-class ReflectionFixtures(object):
+class ReflectionFixtures:
@testing.fixture(
params=[
("engine", True),
class CustomTypeReflectionTest(fixtures.TestBase):
- class CustomType(object):
+ class CustomType:
def __init__(self, arg1=None, arg2=None):
self.arg1 = arg1
self.arg2 = arg2
AnEnum("Baz", 3)
-class ArrayRoundTripTest(object):
+class ArrayRoundTripTest:
__only_on__ = "postgresql"
__backend__ = True
self._assert_data([{r"key \"foo\"": r'value \"bar"\ xyz'}], connection)
def test_orm_round_trip(self, registry):
- class Data(object):
+ class Data:
def __init__(self, name, data):
self.name = name
self.data = data
eq_(data, [(self._data_obj().__class__(empty=True),)])
-class _Int4RangeTests(object):
+class _Int4RangeTests:
_col_type = INT4RANGE
_col_str = "INT4RANGE"
return self.extras().NumericRange(1, 2)
-class _Int8RangeTests(object):
+class _Int8RangeTests:
_col_type = INT8RANGE
_col_str = "INT8RANGE"
)
-class _NumRangeTests(object):
+class _NumRangeTests:
_col_type = NUMRANGE
_col_str = "NUMRANGE"
)
-class _DateRangeTests(object):
+class _DateRangeTests:
_col_type = DATERANGE
_col_str = "DATERANGE"
)
-class _DateTimeRangeTests(object):
+class _DateTimeRangeTests:
_col_type = TSRANGE
_col_str = "TSRANGE"
)
-class _DateTimeTZRangeTests(object):
+class _DateTimeTZRangeTests:
_col_type = TSTZRANGE
_col_str = "TSTZRANGE"
def test_on_conflict_do_update_clauseelem_keys(self, connection):
users = self.tables.users
- class MyElem(object):
+ class MyElem:
def __init__(self, expr):
self.expr = expr
pass
-class Foo(object):
+class Foo:
def __str__(self):
return "foo"
def _dialect(self):
canary = []
- class PoolDialect(object):
+ class PoolDialect:
is_async = False
def do_rollback(self, dbapi_connection):
)
-class ResetFixture(object):
+class ResetFixture:
@testing.fixture()
def reset_agent(self, testing_engine):
engine = testing_engine()
class TestInstrumentDeclarative(fixtures.TestBase):
def test_ok(self):
- class Foo(object):
+ class Foo:
__tablename__ = "foo"
id = sa.Column(sa.Integer, primary_key=True)
with self._expect_warning("as_declarative"):
@legacy_decl.as_declarative()
- class Base(object):
+ class Base:
pass
class Foo(Base):
def test_has_inherited_table(self, registry):
@registry.mapped
- class Foo(object):
+ class Foo:
__tablename__ = "foo"
id = sa.Column(sa.Integer, primary_key=True)
with self._expect_warning("synonym_for"):
@registry.mapped
- class Foo(object):
+ class Foo:
__tablename__ = "foo"
id = sa.Column(sa.Integer, primary_key=True)
@as_declarative()
-class Base(object):
+class Base:
updated_at = Column(Integer)
@reg.as_declarative_base()
-class Base(object):
+class Base:
updated_at = Column(Integer)
@as_declarative()
-class Base(object):
+class Base:
@declared_attr
def __tablename__(self) -> Mapped[str]:
return self.__name__.lower()
pass
-class ObjectCollection(object):
+class ObjectCollection:
def __init__(self):
self.values = list()
)
def _fixture(self, collection_class, is_dict=False):
- class Parent(object):
+ class Parent:
collection = association_proxy("_collection", "child")
- class Child(object):
+ class Child:
def __init__(self, name):
self.name = name
- class Association(object):
+ class Association:
if is_dict:
def __init__(self, key, child):
Column("baz", String(128)),
)
- class Parent(object):
+ class Parent:
foo = association_proxy("child", "foo")
bar = association_proxy(
"child", "bar", creator=lambda v: Child(bar=v)
def __init__(self, name):
self.name = name
- class Child(object):
+ class Child:
def __init__(self, **kw):
for attr in kw:
setattr(self, attr, kw[attr])
Column("name", String(50)),
)
- class A(object):
+ class A:
a2b_name = association_proxy("a2b_single", "name")
b_single = association_proxy("a2b_single", "b")
- class A2B(object):
+ class A2B:
pass
- class B(object):
+ class B:
pass
self.mapper_registry.map_imperatively(
get = Mock()
set_ = Mock()
- class Parent(object):
+ class Parent:
foo = association_proxy(
"child", "foo", getset_factory=lambda cc, parent: (get, set_)
)
- class Child(object):
+ class Child:
def __init__(self, foo):
self.foo = foo
self.assert_(p._children is not None)
-class Parent(object):
+class Parent:
def __init__(self, name):
self.name = name
-class Child(object):
+class Child:
def __init__(self, name):
self.name = name
-class KVChild(object):
+class KVChild:
def __init__(self, name, value):
self.name = name
self.value = value
# assert r2 == {'c1': 'c1', 'c2': 'c2'}
-class PickleKeyFunc(object):
+class PickleKeyFunc:
def __init__(self, name):
self.name = name
Base = declarative_base()
- class Mixin(object):
+ class Mixin:
@declared_attr
def children(cls):
return association_proxy("_children", "value")
def test_resolved_to_correct_class_five(self):
Base = declarative_base()
- class Mixin(object):
+ class Mixin:
children = association_proxy("_children", "value")
class Parent(Mixin, Base):
foo._calc_owner(None, None)
is_(foo.owning_class, None)
- class Bat(object):
+ class Bat:
foo = association_proxy("x", "y")
Bat.foo
is_(Bat.foo.owning_class, Bat)
-class ScalarRemoveTest(object):
+class ScalarRemoveTest:
useobject = None
cascade_scalar_deletes = None
uselist = None
eq_(assoc.info, {})
def test_via_cls(self):
- class Foob(object):
+ class Foob:
assoc = association_proxy("a", "b")
eq_(Foob.assoc.info, {})
def test_get_pk_w_null(self):
"""test the re-implementation of logic to do get with IS NULL."""
- class AddressUser(object):
+ class AddressUser:
pass
self.mapper_registry.map_imperatively(
instrumentation.instrumentation_finders.extend(pristine)
-class _ExtBase(object):
+class _ExtBase:
@classmethod
def teardown_test_class(cls):
instrumentation._reinstall_default_lookups()
return get
- class MyClass(object):
+ class MyClass:
__sa_instrumentation_manager__ = MyClassState
assert attributes.manager_of_class(MyClass) is None
def setup_test_class(cls):
global MyBaseClass, MyClass
- class MyBaseClass(object):
+ class MyBaseClass:
__sa_instrumentation_manager__ = (
instrumentation.InstrumentationManager
)
- class MyClass(object):
+ class MyClass:
# This proves that a staticmethod will work here; don't
# flatten this back to a class assignment!
def test_alternate_finders(self):
"""Ensure the generic finder front-end deals with edge cases."""
- class Unknown(object):
+ class Unknown:
pass
class Known(MyBaseClass):
class FinderTest(_ExtBase, fixtures.ORMTest):
def test_standard(self):
- class A(object):
+ class A:
pass
register_class(A)
eq_(type(manager_of_class(A)), instrumentation.ClassManager)
def test_nativeext_interfaceexact(self):
- class A(object):
+ class A:
__sa_instrumentation_manager__ = (
instrumentation.InstrumentationManager
)
class Mine(instrumentation.ClassManager):
pass
- class A(object):
+ class A:
__sa_instrumentation_manager__ = Mine
register_class(A)
class Mine(instrumentation.ClassManager):
pass
- class A(object):
+ class A:
pass
def find(cls):
@modifies_instrumentation_finders
def test_customfinder_pass(self):
- class A(object):
+ class A:
pass
def find(cls):
class InstrumentationCollisionTest(_ExtBase, fixtures.ORMTest):
def test_none(self):
- class A(object):
+ class A:
pass
register_class(A)
def mgr_factory(cls):
return instrumentation.ClassManager(cls)
- class B(object):
+ class B:
__sa_instrumentation_manager__ = staticmethod(mgr_factory)
register_class(B)
- class C(object):
+ class C:
__sa_instrumentation_manager__ = instrumentation.ClassManager
register_class(C)
def test_single_down(self):
- class A(object):
+ class A:
pass
register_class(A)
)
def test_single_up(self):
- class A(object):
+ class A:
pass
# delay registration
def mgr_factory(cls):
return instrumentation.ClassManager(cls)
- class A(object):
+ class A:
pass
class B1(A):
class B2(A):
__sa_instrumentation_manager__ = staticmethod(mgr_factory)
- class C(object):
+ class C:
pass
assert_raises_message(
def mgr_factory(cls):
return instrumentation.ClassManager(cls)
- class A(object):
+ class A:
pass
class B1(A):
class B2(A):
__sa_instrumentation_manager__ = staticmethod(mgr_factory)
- class C(object):
+ class C:
pass
register_class(B2)
def mgr_factory(cls):
return instrumentation.ClassManager(cls)
- class A(object):
+ class A:
pass
class B1(A):
class B2(A):
__sa_instrumentation_manager__ = staticmethod(mgr_factory)
- class C(object):
+ class C:
pass
register_class(C)
0, lambda cls: MyClassManager
)
- class A(object):
+ class A:
pass
register_class(A)
from sqlalchemy.testing.engines import testing_reaper
-class ShardTest(object):
+class ShardTest:
__skip_if__ = (lambda: util.win32,)
__requires__ = ("sqlite",)
def setup_mappers(cls):
global WeatherLocation, Report
- class WeatherLocation(object):
+ class WeatherLocation:
def __init__(self, continent, city):
self.continent = continent
self.city = city
- class Report(object):
+ class Report:
def __init__(self, temperature, id_=None):
self.temperature = temperature
if id_:
"""test :ticket:`6215`"""
Base = declarative_base()
- class SomeMixin(object):
+ class SomeMixin:
@hybrid.hybrid_property
def same_name(self):
return self.id
for currency_from, rate in zip(symbols, values)
)
- class Amount(object):
+ class Amount:
def __init__(self, amount, currency):
self.currency = currency
self.amount = amount
pass
-class FooWithEq(object):
+class FooWithEq:
def __init__(self, **kw):
for k in kw:
setattr(self, k, kw[k])
return value
-class _MutableDictTestFixture(object):
+class _MutableDictTestFixture:
@classmethod
def _type_fixture(cls):
return MutableDict
eq_(f1.data, {"a": "b"})
-class _MutableListTestFixture(object):
+class _MutableListTestFixture:
@classmethod
def _type_fixture(cls):
return MutableList
assert isinstance(obj, MutableList)
-class _MutableSetTestFixture(object):
+class _MutableSetTestFixture:
@classmethod
def _type_fixture(cls):
return MutableSet
eq_(f1.data, set([1, 2]))
-class _MutableNoHashFixture(object):
+class _MutableNoHashFixture:
@testing.fixture(autouse=True, scope="class")
def set_class(self):
global Foo
Base = declarative_base(metadata=metadata)
- class Mixin(object):
+ class Mixin:
data = Column(MutableList.as_mutable(ARRAY(Integer)))
class Foo(Mixin, Base):
eq_(type(f1.data), self._type_fixture())
-class _CompositeTestBase(object):
+class _CompositeTestBase:
@classmethod
def define_tables(cls, metadata):
Table(
Column("text", String(128)),
)
- class Slide(object):
+ class Slide:
def __init__(self, name):
self.name = name
def __repr__(self):
return '<Slide "%s">' % self.name
- class Bullet(object):
+ class Bullet:
def __init__(self, text):
self.text = text
eq_([b.position for b in s1.bullets], [0, 1, 2])
def test_funky_ordering(self):
- class Pos(object):
+ class Pos:
def __init__(self):
self.position = None
self.assert_(copy.__dict__ == olist.__dict__)
-class DummyItem(object):
+class DummyItem:
def __init__(self, order=None):
self.order = order
return CannedResults(self)
-class CannedResults(object):
+class CannedResults:
"""Built on demand, instances use mappers in effect at time of call."""
def __init__(self, test):
def test_dispose_attrs(self):
reg = registry()
- class Foo(object):
+ class Foo:
__tablename__ = "some_table"
id = Column(Integer, primary_key=True)
)
def test_deferred_reflection_default_error(self):
- class MyExt(object):
+ class MyExt:
@classmethod
def prepare(cls):
"sample prepare method"
x = sa.sql.expression.column(Integer)
y = Column(Integer)
- class MyMixin(object):
+ class MyMixin:
x = sa.sql.expression.column(Integer)
y = Column(Integer)
r"non-schema 'sqlalchemy.sql.column\(\)' object; "
):
- class MyMixin2(object):
+ class MyMixin2:
@declared_attr
def x(cls):
return sa.sql.expression.column(Integer)
)
def test_custom_base(self):
- class MyBase(object):
+ class MyBase:
def foobar(self):
return "foobar"
def test_table_cls_attribute_return_none(self):
from sqlalchemy.schema import Column, PrimaryKeyConstraint
- class AutoTable(object):
+ class AutoTable:
@declared_attr
def __tablename__(cls):
return cls.__name__
)
def test_cls_docstring(self):
- class MyBase(object):
+ class MyBase:
"""MyBase Docstring"""
Base = declarative_base(cls=MyBase)
from sqlalchemy.testing.util import gc_collect
-class MockClass(object):
+class MockClass:
def __init__(self, base, name):
self._sa_class_manager = mock.Mock(registry=base)
tokens = name.split(".")
self.metadata = MetaData()
-class MockProp(object):
+class MockProp:
parent = "some_parent"
reg = registry(_bind=testing.db)
@reg.mapped
- class User(object):
+ class User:
__tablename__ = "user"
id = Column(Integer, primary_key=True)
discriminator = Column("type", String(50))
__mapper_args__ = {"polymorphic_on": discriminator}
- class MyMixin(object):
+ class MyMixin:
pass
assert class_mapper(Engineer).inherits is class_mapper(Person)
def test_intermediate_abstract_class_on_classical(self):
- class Person(object):
+ class Person:
pass
person_table = Table(
eq_(set(class_mapper(Manager).class_manager), {"id", "kind"})
def test_intermediate_unmapped_class_on_classical(self):
- class Person(object):
+ class Person:
pass
person_table = Table(
eq_(set(class_mapper(Manager).class_manager), {"id", "kind"})
def test_class_w_invalid_multiple_bases(self):
- class Person(object):
+ class Person:
pass
person_table = Table(
class DeclarativeMixinTest(DeclarativeTestBase):
def test_simple_wbase(self):
- class MyMixin(object):
+ class MyMixin:
id = Column(
Integer, primary_key=True, test_needs_autoincrement=True
eq_(obj.foo(), "bar1")
def test_simple_wdecorator(self):
- class MyMixin(object):
+ class MyMixin:
id = Column(
Integer, primary_key=True, test_needs_autoincrement=True
eq_(obj.foo(), "bar1")
def test_unique_column(self):
- class MyMixin(object):
+ class MyMixin:
id = Column(Integer, primary_key=True)
value = Column(String, unique=True)
def test_mixin_overrides_wbase(self):
"""test a mixin that overrides a column on a superclass."""
- class MixinA(object):
+ class MixinA:
foo = Column(String(50))
class MixinB(MixinA):
def test_mixin_overrides_wdecorator(self):
"""test a mixin that overrides a column on a superclass."""
- class MixinA(object):
+ class MixinA:
foo = Column(String(50))
class MixinB(MixinA):
reg = registry()
- class B1(object):
+ class B1:
metadata = m1
- class B2(object):
+ class B2:
metadata = m2
def fullname(self):
username = Column(String)
@reg.mapped
- class BUser(object):
+ class BUser:
__tablename__ = "user"
id = Column(Integer, primary_key=True)
eq_(MyModel.__table__.name, "mymodel")
def test_classproperty_still_works(self):
- class MyMixin(object):
+ class MyMixin:
@classproperty
def __tablename__(cls):
return cls.__name__.lower()
"""
- class MyMixin(object):
+ class MyMixin:
foo = Column("foo", Integer)
bar = Column("bar_newname", Integer)
__tablename__ = "person"
id = Column(Integer, primary_key=True)
- class Mixin(object):
+ class Mixin:
@declared_attr
def target_id(cls):
return cls.__table__.c.get(
"""
- class MyMixin(object):
+ class MyMixin:
foo = Column("foo", Integer)
bar = Column("bar_newname", Integer)
def test_declare_first_mixin(self):
canary = mock.Mock()
- class MyMixin(object):
+ class MyMixin:
@classmethod
def __declare_first__(cls):
canary.declare_first__(cls)
def test_declare_first_base(self):
canary = mock.Mock()
- class MyMixin(object):
+ class MyMixin:
@classmethod
def __declare_first__(cls):
canary.declare_first__(cls)
def test_mapper_args_custom_base(self):
"""test the @declared_attr approach from a custom base."""
- class Base(object):
+ class Base:
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
eq_(Joined.__table__.kwargs, {"mysql_engine": "InnoDB"})
def test_col_copy_vs_declared_attr_joined_propagation(self):
- class Mixin(object):
+ class Mixin:
a = Column(Integer)
@declared_attr
assert "b" not in B.__table__.c
def test_col_copy_vs_declared_attr_joined_propagation_newname(self):
- class Mixin(object):
+ class Mixin:
a = Column("a1", Integer)
@declared_attr
assert "b1" not in B.__table__.c
def test_col_copy_vs_declared_attr_single_propagation(self):
- class Mixin(object):
+ class Mixin:
a = Column(Integer)
@declared_attr
assert model_col.type.__class__ is Integer
def test_mixin_column_ordering(self):
- class Foo(object):
+ class Foo:
col1 = Column(Integer)
col3 = Column(Integer)
- class Bar(object):
+ class Bar:
col2 = Column(Integer)
col4 = Column(Integer)
)
def test_honor_class_mro_one(self):
- class HasXMixin(object):
+ class HasXMixin:
@declared_attr
def x(self):
return Column(Integer)
assert "x" not in Child.__table__.c
def test_honor_class_mro_two(self):
- class HasXMixin(object):
+ class HasXMixin:
@declared_attr
def x(self):
return Column(Integer)
assert C().x() == "hi"
def test_arbitrary_attrs_one(self):
- class HasMixin(object):
+ class HasMixin:
@declared_attr
def some_attr(cls):
return cls.__name__ + "SOME ATTR"
def __init__(self, filter_, **kw):
self.filter = filter_
- class FilterMixin(object):
+ class FilterMixin:
@declared_attr
def _filters(cls):
return relationship(
DeclarativeTestBase, testing.AssertsCompiledSQL
):
def test_column_property(self):
- class MyMixin(object):
+ class MyMixin:
@declared_attr
def prop_hoho(cls):
return column_property(Column("prop", String(50)))
"""
- class MyMixin(object):
+ class MyMixin:
@declared_attr
def type_(cls):
"""this is a document."""
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import inspect
- class Mixin(object):
+ class Mixin:
@hybrid_property
def hp1(cls):
return 42
def test_correct_for_proxies_doesnt_impact_synonyms(self):
from sqlalchemy import inspect
- class Mixin(object):
+ class Mixin:
@declared_attr
def data_syn(cls):
return synonym("data")
)
def test_column_in_mapper_args(self):
- class MyMixin(object):
+ class MyMixin:
@declared_attr
def type_(cls):
return Column(String(50))
assert col.table is not None
def test_column_in_mapper_args_used_multiple_times(self):
- class MyMixin(object):
+ class MyMixin:
version_id = Column(Integer)
__mapper_args__ = {"version_id_col": version_id}
)
def test_deferred(self):
- class MyMixin(object):
+ class MyMixin:
@declared_attr
def data(cls):
return deferred(Column("data", String(50)))
assert "data" in d1.__dict__
def _test_relationship(self, usestring):
- class RefTargetMixin(object):
+ class RefTargetMixin:
@declared_attr
def target_id(cls):
return Column("target_id", ForeignKey("target.id"))
def test_singleton_behavior_within_decl(self):
counter = mock.Mock()
- class Mixin(object):
+ class Mixin:
@declared_attr
def my_prop(cls):
counter(cls)
def test_singleton_gc(self):
counter = mock.Mock()
- class Mixin(object):
+ class Mixin:
@declared_attr
def my_prop(cls):
counter(cls.__name__)
)
def test_can_we_access_the_mixin_straight(self):
- class Mixin(object):
+ class Mixin:
@declared_attr
def my_prop(cls):
return Column("x", Integer)
)
def test_can_we_access_the_mixin_straight_special_names(self):
- class Mixin(object):
+ class Mixin:
@declared_attr
def __table_args__(cls):
return (1, 2, 3)
def test_non_decl_access(self):
counter = mock.Mock()
- class Mixin(object):
+ class Mixin:
@declared_attr
def __tablename__(cls):
counter(cls)
def test_property_noncascade(self):
counter = mock.Mock()
- class Mixin(object):
+ class Mixin:
@declared_attr
def my_prop(cls):
counter(cls)
def test_property_cascade_mixin(self):
counter = mock.Mock()
- class Mixin(object):
+ class Mixin:
@declared_attr.cascading
def my_prop(cls):
counter(cls)
def test_property_cascade_mixin_override(self):
counter = mock.Mock()
- class Mixin(object):
+ class Mixin:
@declared_attr.cascading
def my_prop(cls):
counter(cls)
eq_(counter.mock_calls, [mock.call(A), mock.call(B)])
def test_warn_cascading_used_w_tablename(self):
- class Mixin(object):
+ class Mixin:
@declared_attr.cascading
def __tablename__(cls):
return "foo"
asserted = collections.defaultdict(set)
- class Mixin(object):
+ class Mixin:
@declared_attr.cascading
def my_attr(cls):
if has_inherited_table(cls):
def test_column_pre_map(self):
counter = mock.Mock()
- class Mixin(object):
+ class Mixin:
@declared_attr
def my_col(cls):
counter(cls)
counter = mock.Mock()
- class HasAddressCount(object):
+ class HasAddressCount:
id = Column(Integer, primary_key=True)
@declared_attr
def test_implicit_abstract_viadecorator(self):
@mapper_registry.mapped
- class A(object):
+ class A:
__tablename__ = "a"
id = Column(Integer, primary_key=True)
if usedata:
- class Data(object):
+ class Data:
def __init__(self, data):
self.data = data
how to alias the primaryjoin to the polymorphic union ?"""
# class definitions
- class Person(object):
+ class Person:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
self.longer_status,
)
- class Car(object):
+ class Car:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
"""test parent object with child relationship to an inheriting mapper,
using eager loads, works when there are no child objects present"""
- class Person(object):
+ class Person:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
self.longer_status,
)
- class Car(object):
+ class Car:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
"""test that lazy load clause to a polymorphic child mapper generates
correctly [ticket:493]"""
- class PersistentObject(object):
+ class PersistentObject:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
)
def test_threelevels(self):
- class Employee(object):
+ class Employee:
def set(me, **kargs):
for k, v in kargs.items():
setattr(me, k, v)
aren't attempted to be matched to the target polymorphic
selectable"""
- class BaseItem(object):
+ class BaseItem:
pass
class Item(BaseItem):
pass
- class Collection(object):
+ class Collection:
pass
item_join = polymorphic_union(
"""test that the primary_key attribute is propagated to the
polymorphic mapper"""
- class T1(object):
+ class T1:
pass
class T2(T1):
"""test that a composite primary key attribute formed by a join
is "collapsed" into its minimal columns"""
- class T1(object):
+ class T1:
pass
class T2(T1):
)
def test_basic(self):
- class Foo(object):
+ class Foo:
def __init__(self, data=None):
self.data = data
)
def test_false_on_sub(self):
- class Foo(object):
+ class Foo:
pass
class Bar(Foo):
assert isinstance(sess.query(Foo).one(), Bar)
def test_false_on_base(self):
- class Ding(object):
+ class Ding:
pass
class Bat(Ding):
self.tables.user_roles,
)
- class User(object):
+ class User:
pass
- class Role(object):
+ class Role:
pass
class Admin(User):
self.tables.user_roles,
)
- class User(object):
+ class User:
def __init__(self, email=None, password=None):
self.email = email
self.password = password
- class Role(object):
+ class Role:
def __init__(self, description=None):
self.description = description
Column("person_id", Integer, ForeignKey("persons.id")),
)
- class Person(object):
+ class Person:
def __init__(self, name):
self.name = name
j1 = testing.resolve_lambda(j1, **locals())
j2 = testing.resolve_lambda(j2, **locals())
- class A(object):
+ class A:
def __init__(self, **kwargs):
for key, value in list(kwargs.items()):
setattr(self, key, value)
def test_plain(self):
# control case
- class Base(object):
+ class Base:
pass
class Sub(Base):
# in particular, here we do a "manual" version of
# what we'd like the mapper to do.
- class Base(object):
+ class Base:
pass
class Sub(Base):
assert sess.get(Sub, 10) is s1
def test_override_onlyinparent(self):
- class Base(object):
+ class Base:
pass
class Sub(Base):
# this is originally [ticket:1111].
# the pattern here is now disallowed by [ticket:1892]
- class Base(object):
+ class Base:
pass
class Sub(Base):
assert_raises(sa_exc.InvalidRequestError, go)
def test_pk_fk_different(self):
- class Base(object):
+ class Base:
pass
class Sub(Base):
"""test that descriptors prevent inheritance from propagating
properties to subclasses."""
- class Base(object):
+ class Base:
pass
class Sub(Base):
"""test that descriptors prevent inheritance from propagating
properties to subclasses."""
- class MyDesc(object):
+ class MyDesc:
def __get__(self, instance, owner):
if instance is None:
return self
return "im the data"
- class Base(object):
+ class Base:
pass
class Sub(Base):
assert sess.query(Sub).one().data == "im the data"
def test_sub_columns_over_base_descriptors(self):
- class Base(object):
+ class Base:
@property
def subdata(self):
return "this is base"
assert sess.get(Sub, s1.base_id).subdata == "this is sub"
def test_base_descriptors_over_base_cols(self):
- class Base(object):
+ class Base:
@property
def data(self):
return "this is base"
class WithComp(Base):
pass
- class Comp(object):
+ class Comp:
def __init__(self, a, b):
self.a = a
self.b = b
def test_warning_on_sub(self):
parent, child = self._fixture()
- class P(object):
+ class P:
pass
class C(P):
def test_no_warning_with_explicit(self):
parent, child = self._fixture()
- class P(object):
+ class P:
pass
class C(P):
Column("owner_id", Integer, ForeignKey("owner.owner_id")),
)
- class Base(object):
+ class Base:
pass
class Derived(Base):
)
Table("order", m, Column("id", Integer, primary_key=True))
- class Base(object):
+ class Base:
pass
class Derived(Base):
"derived", metadata, Column("id", Integer, primary_key=True)
)
- class Base(object):
+ class Base:
pass
class Derived(Base):
Column("id", Integer, ForeignKey("base.id"), primary_key=True),
)
- class Base(object):
+ class Base:
pass
class Derived(Base):
Column("id", Integer, ForeignKey("base.id"), primary_key=True),
)
- class Base(object):
+ class Base:
pass
class Derived(Base):
Column("id", Integer, ForeignKey("base.q"), primary_key=True),
)
- class Base(object):
+ class Base:
pass
class Derived(Base):
def test_pk_as_discriminator(self):
parents, children = self.tables.parents, self.tables.children
- class Parent(object):
+ class Parent:
def __init__(self, name=None):
self.name = name
- class Child(object):
+ class Child:
def __init__(self, name=None):
self.name = name
)
def test_name_conflict(self):
- class Content(object):
+ class Content:
pass
class Foo(Content):
"pjoin",
)
- class Location(object):
+ class Location:
pass
class Refugee(Location):
)
def test_basic(self):
- class Principal(object):
+ class Principal:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
)
def test_get(self):
- class Foo(object):
+ class Foo:
def __init__(self, data=None):
self.data = data
assert sess.get(Bar, b.id).id == b.id
def test_basic(self):
- class Foo(object):
+ class Foo:
def __init__(self, data=None):
self.data = data
)
def test_basic(self):
- class Foo(object):
+ class Foo:
def __init__(self, data=None):
self.data = data
eq_(found, compare)
def test_advanced(self):
- class Foo(object):
+ class Foo:
def __init__(self, data=None):
self.data = data
join = table1.outerjoin(table2).outerjoin(table3).alias("pjoin")
# join = None
- class Table1(object):
+ class Table1:
def __init__(self, name, data=None):
self.name = name
if data is not None:
class Table3(Table1):
pass
- class Data(object):
+ class Data:
def __init__(self, data):
self.data = data
from ._poly_fixtures import Person
-class BaseAndSubFixture(object):
+class BaseAndSubFixture:
use_options = False
@classmethod
Column("size", Integer, default=0),
)
- class Product(object):
+ class Product:
def __init__(self, name, mark=""):
self.name = name
self.mark = mark
)
)
- class SpecLine(object):
+ class SpecLine:
def __init__(self, leader=None, follower=None, quantity=1):
self.leader = leader
self.follower = follower
repr(self.follower),
)
- class Document(object):
+ class Document:
def __init__(self, name, data=None):
self.name = name
self.data = data
class AttributeImplAPITest(fixtures.MappedTest):
def _scalar_obj_fixture(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
instrumentation.register_class(A)
return A, B
def _collection_obj_fixture(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
instrumentation.register_class(A)
def setup_test(self):
global MyTest, MyTest2
- class MyTest(object):
+ class MyTest:
pass
- class MyTest2(object):
+ class MyTest2:
pass
def teardown_test(self):
MyTest, MyTest2 = None, None
def test_basic(self):
- class User(object):
+ class User:
pass
instrumentation.register_class(User)
"""test that InstanceState always has a dict, even after host
object gc'ed."""
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
@testing.requires.predictable_gc
def test_object_dereferenced_error(self):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
def __init__(self):
gc_collect()
)
def test_unmapped_instance_raises(self):
- class User(object):
+ class User:
pass
instrumentation.register_class(User)
User, "user_name", uselist=False, useobject=False
)
- class Blog(object):
+ class Blog:
name = User.user_name
def go():
)
def test_del_scalar_nonobject(self):
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
)
def test_del_scalar_object(self):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
)
def test_del_collection_object(self):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
eq_(f1.b, [])
def test_deferred(self):
- class Foo(object):
+ class Foo:
pass
data = {"a": "this is a", "b": 12}
eq_(m2.b, 12)
def test_list(self):
- class User(object):
+ class User:
pass
- class Address(object):
+ class Address:
pass
instrumentation.register_class(User)
"""
- class Post(object):
+ class Post:
pass
- class Blog(object):
+ class Blog:
pass
instrumentation.register_class(Post)
assert attributes.has_parent(Post, b2, "blog")
def test_illegal_trackparent(self):
- class Post(object):
+ class Post:
pass
- class Blog(object):
+ class Blog:
pass
instrumentation.register_class(Post)
def test_inheritance(self):
"""tests that attributes are polymorphic"""
- class Foo(object):
+ class Foo:
pass
class Bar(Foo):
def test_no_double_state(self):
states = set()
- class Foo(object):
+ class Foo:
def __init__(self):
states.add(attributes.instance_state(self))
managed attributes of an object, if the object is of a
descendant class with managed attributes in the parent class"""
- class Foo(object):
+ class Foo:
pass
class Bar(Foo):
pass
- class Element(object):
+ class Element:
_state = True
instrumentation.register_class(Foo)
)
def test_parenttrack(self):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
from ones implementing zope.interface.Interface). This is a
simple regression test to prevent that defect."""
- class des(object):
+ class des:
def __get__(self, instance, owner):
raise AttributeError("fake attribute")
- class Foo(object):
+ class Foo:
A = des()
instrumentation.register_class(Foo)
instrumentation.unregister_class(Foo)
def test_collectionclasses(self):
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
assert isinstance(Foo().collection, MyDict)
attributes.unregister_attribute(Foo, "collection")
- class MyColl(object):
+ class MyColl:
pass
try:
"collection class"
)
- class MyColl(object):
+ class MyColl:
@collection.iterator
def __iter__(self):
return iter([])
assert False
def test_last_known_tracking(self):
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
class GetNoValueTest(fixtures.ORMTest):
def _fixture(self, expected):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
def lazy_callable(state, passive):
class UtilTest(fixtures.ORMTest):
def test_helpers(self):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
"""test that set_committed_value->None to a uselist generates an
empty list"""
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
eq_(f1.col_set, set())
def test_initiator_arg(self):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
class BackrefTest(fixtures.ORMTest):
def test_m2m(self):
- class Student(object):
+ class Student:
pass
- class Course(object):
+ class Course:
pass
instrumentation.register_class(Student)
self.assert_(c.students == [s2, s3])
def test_o2m(self):
- class Post(object):
+ class Post:
pass
- class Blog(object):
+ class Blog:
pass
instrumentation.register_class(Post)
del p5.blog
def test_o2o(self):
- class Port(object):
+ class Port:
pass
- class Jack(object):
+ class Jack:
pass
instrumentation.register_class(Port)
"""
- class Parent(object):
+ class Parent:
pass
- class Child(object):
+ class Child:
pass
class SubChild(Child):
c2.parent = p1
def test_symmetric_o2m_inheritance(self):
- class Parent(object):
+ class Parent:
pass
class SubParent(Parent):
pass
- class Child(object):
+ class Child:
pass
p_token = object()
)
def _scalar_fixture(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
- class C(object):
+ class C:
pass
instrumentation.register_class(A)
return A, B, C
def _collection_fixture(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
- class C(object):
+ class C:
pass
instrumentation.register_class(A)
return A, B, C
def _broken_collection_fixture(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
instrumentation.register_class(A)
class PendingBackrefTest(fixtures.ORMTest):
def _fixture(self):
- class Post(object):
+ class Post:
def __init__(self, name):
self.name = name
def __eq__(self, other):
return other is not None and other.name == self.name
- class Blog(object):
+ class Blog:
def __init__(self, name):
self.name = name
def test_receive_changes(self):
"""test that Listeners can mutate the given value."""
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
def append(state, child, initiator):
def test_named(self):
canary = Mock()
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
)
def test_collection_link_events(self):
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
"""
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
def test_flag_modified(self):
canary = Mock()
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
def test_none_init_scalar(self):
canary = Mock()
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
def test_none_init_object(self):
canary = Mock()
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
def test_none_init_collection(self):
canary = Mock()
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
instrumentation.register_class(Foo)
canary = []
def make_a():
- class A(object):
+ class A:
pass
classes[0] = A
classes[2] = C
def make_d():
- class D(object):
+ class D:
pass
classes[3] = D
class CollectionInitTest(fixtures.TestBase):
def setup_test(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
self.A = A
class TestUnlink(fixtures.TestBase):
def setup_test(self):
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
self.A = A
from sqlalchemy import Column, Integer, String
- class Foo(object):
+ class Foo:
id = Column(Integer)
name = Column(String)
def test_loader_criteria_bound_param_thing(self):
from sqlalchemy import Column, Integer
- class Foo(object):
+ class Foo:
id = Column(Integer)
def go(param):
r2_present,
detach_event=True,
):
- class Core(object):
+ class Core:
pass
- class RelatedOne(object):
+ class RelatedOne:
def __init__(self, cores):
self.cores = cores
- class RelatedTwo(object):
+ class RelatedTwo:
def __init__(self, cores):
self.cores = cores
def cascade_fixture(self, registry):
def go(collection_class):
@registry.mapped
- class A(object):
+ class A:
__tablename__ = "a"
id = Column(Integer, primary_key=True)
)
@registry.mapped
- class B(object):
+ class B:
__tablename__ = "b_"
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey("a.id"))
from sqlalchemy.testing.schema import Table
-class Canary(object):
+class Canary:
def __init__(self):
self.data = set()
self.added = set()
return value
-class OrderedDictFixture(object):
+class OrderedDictFixture:
@testing.fixture
def ordered_dict_mro(self):
return type("ordered", (collections.MappedCollection,), {})
class CollectionsTest(OrderedDictFixture, fixtures.ORMTest):
- class Entity(object):
+ class Entity:
def __init__(self, a=None, b=None, c=None):
self.a = a
self.b = b
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
canary = Canary()
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
canary = Canary()
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
canary = Canary()
# or __delslice__ methods. The __setitem__
# and __delitem__ must therefore accept
# slice objects (i.e. as in py3k)
- class ListLike(object):
+ class ListLike:
def __init__(self):
self.data = list()
self.assert_(getattr(MyList, "_sa_instrumented") == id(MyList))
def test_list_duck(self):
- class ListLike(object):
+ class ListLike:
def __init__(self):
self.data = list()
self.assert_(getattr(ListLike, "_sa_instrumented") == id(ListLike))
def test_list_emulates(self):
- class ListIsh(object):
+ class ListIsh:
__emulates__ = list
def __init__(self):
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
canary = Canary()
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
canary = Canary()
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
canary = Canary()
self.assert_(getattr(MySet, "_sa_instrumented") == id(MySet))
def test_set_duck(self):
- class SetLike(object):
+ class SetLike:
def __init__(self):
self.data = set()
self.assert_(getattr(SetLike, "_sa_instrumented") == id(SetLike))
def test_set_emulates(self):
- class SetIsh(object):
+ class SetIsh:
__emulates__ = set
def __init__(self):
if creator is None:
creator = self.dictable_entity
- class Foo(object):
+ class Foo:
pass
canary = Canary()
if creator is None:
creator = self.dictable_entity
- class Foo(object):
+ class Foo:
pass
canary = Canary()
if creator is None:
creator = self.dictable_entity
- class Foo(object):
+ class Foo:
pass
canary = Canary()
self.assert_(getattr(MyOrdered, "_sa_instrumented") == id(MyOrdered))
def test_dict_duck(self):
- class DictLike(object):
+ class DictLike:
def __init__(self):
self.data = dict()
self.assert_(getattr(DictLike, "_sa_instrumented") == id(DictLike))
def test_dict_emulates(self):
- class DictIsh(object):
+ class DictIsh:
__emulates__ = dict
def __init__(self):
if creator is None:
creator = self.entity_maker
- class Foo(object):
+ class Foo:
pass
canary = Canary()
assert_eq()
def test_object_duck(self):
- class MyCollection(object):
+ class MyCollection:
def __init__(self):
self.data = set()
)
def test_object_emulates(self):
- class MyCollection2(object):
+ class MyCollection2:
__emulates__ = None
def __init__(self):
)
def test_recipes(self):
- class Custom(object):
+ class Custom:
def __init__(self):
self.data = []
def __iter__(self):
return iter(self.data)
- class Foo(object):
+ class Foo:
pass
canary = Canary()
self.assert_(dr3 is cr3)
def test_lifecycle(self):
- class Foo(object):
+ class Foo:
pass
canary = Canary()
class MyList(list):
pass
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
self.mapper_registry.map_imperatively(
self.tables.sometable,
)
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
self.mapper_registry.map_imperatively(
self.tables.sometable,
)
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
class AppenderDict(dict):
self.tables.sometable,
)
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
def __init__(self, data):
self.data = data
self._test_list(list)
def test_list_no_setslice(self):
- class ListLike(object):
+ class ListLike:
def __init__(self):
self.data = list()
self.tables.sometable,
)
- class Parent(object):
+ class Parent:
pass
- class Child(object):
+ class Child:
pass
self.mapper_registry.map_imperatively(
self.tables.sometable,
)
- class Parent(object):
+ class Parent:
pass
- class Child(object):
+ class Child:
pass
- class MyCollection(object):
+ class MyCollection:
def __init__(self):
self.data = []
class InstrumentationTest(fixtures.ORMTest):
def test_uncooperative_descriptor_in_sweep(self):
- class DoNotTouch(object):
+ class DoNotTouch:
def __get__(self, obj, owner):
raise AttributeError
collections._instrument_class(Touchy)
def test_referenced_by_owner(self):
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
),
)
- class Order(object):
+ class Order:
pass
- class Employee(object):
+ class Employee:
pass
- class Product(object):
+ class Product:
pass
- class OrderProduct(object):
+ class OrderProduct:
pass
order_join = order.select().alias("pjoin")
),
)
- class Order(object):
+ class Order:
pass
- class Product(object):
+ class Product:
pass
- class OrderProduct(object):
+ class OrderProduct:
pass
order_join = order.select().alias("pjoin")
metadata.create_all(connection)
connection.execute(node_table.insert(), dict(node_id=1, node_index=5))
- class Node(object):
+ class Node:
pass
- class NodeName(object):
+ class NodeName:
pass
- class Host(object):
+ class Host:
pass
self.mapper_registry.map_imperatively(Node, node_table)
Column("a_id", Integer, ForeignKey("a.id")),
)
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
self.mapper_registry.map_imperatively(
Column("a_id", Integer, ForeignKey("a.id")),
)
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
class C(B):
)
def _fixture(self):
- class AB(object):
+ class AB:
def __init__(self, a, b, cd):
self.a = a
self.b = b
def __ne__(self, other):
return not self.__eq__(other)
- class CD(object):
+ class CD:
def __init__(self, c, d):
self.c = c
self.d = d
def __ne__(self, other):
return not self.__eq__(other)
- class Thing(object):
+ class Thing:
def __init__(self, ab):
self.ab = ab
)
def test_col_prop_builtin_function(self):
- class Foo(object):
+ class Foo:
pass
self.mapper_registry.map_imperatively(
__dialect__ = "default"
def test_deferred_scalar_loader_name_change(self):
- class Foo(object):
+ class Foo:
pass
def myloader(*arg, **kw):
for key, value in dictlike.items():
yield value + 5
- class Foo(object):
+ class Foo:
pass
instrumentation.register_class(Foo)
"AttributeEvents"
):
- class Base(object):
+ class Base:
@collection.iterator
def base_iterate(self, x):
return "base_iterate"
def test_illegal_non_primary_3(self):
users, addresses = self.tables.users, self.tables.addresses
- class Base(object):
+ class Base:
pass
class Sub(Base):
def test_illegal_non_primary_3_legacy(self):
users, addresses = self.tables.users, self.tables.addresses
- class Base(object):
+ class Base:
pass
class Sub(Base):
):
@as_declarative(bind=testing.db)
- class Base(object):
+ class Base:
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
self.tables.t1,
)
- class T1(object):
+ class T1:
pass
- class T2(object):
+ class T2:
pass
self.mapper_registry.map_imperatively(
# building a totally separate metadata /mapping here
# because we need to control if the MetaData is bound or not
- class User(object):
+ class User:
pass
m = MetaData()
def cascade_fixture(self, registry):
def go(collection_class):
@registry.mapped
- class A(object):
+ class A:
__tablename__ = "a"
id = Column(Integer, primary_key=True)
)
@registry.mapped
- class B(object):
+ class B:
__tablename__ = "b_"
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey("a.id"))
from test.orm import _fixtures
-class _DynamicFixture(object):
+class _DynamicFixture:
def _user_address_fixture(self, addresses_args={}):
users, Address, addresses, User = (
self.tables.users,
)
item_keywords = self.tables.item_keywords
- class ItemKeyword(object):
+ class ItemKeyword:
pass
self.mapper_registry.map_imperatively(
def test_basic(self):
t2, t1 = self.tables.t2, self.tables.t1
- class T(object):
+ class T:
pass
class SubT(T):
pass
- class T2(object):
+ class T2:
pass
class SubT2(T2):
from test.orm import _fixtures
-class _RemoveListeners(object):
+class _RemoveListeners:
@testing.fixture(autouse=True)
def _remove_listeners(self):
yield
canary = []
- class A(object):
+ class A:
pass
class B(A):
def test_init_failure_hook(self):
users = self.tables.users
- class Thing(object):
+ class Thing:
def __init__(self, **kw):
if kw.get("fail"):
raise Exception("failure")
def test_instrument_class_precedes_class_instrumentation(self):
users = self.tables.users
- class MyClass(object):
+ class MyClass:
pass
class MySubClass(MyClass):
def test_isolation_instrument_event(self):
User = self.classes.User
- class Bar(object):
+ class Bar:
pass
canary = []
@testing.requires.predictable_gc
def test_instrument_event_auto_remove(self):
- class Bar(object):
+ class Bar:
pass
dispatch = instrumentation._instrumentation_factory.dispatch
def test_unmapped_listen(self):
users = self.tables.users
- class Foo(object):
+ class Foo:
pass
fn = Mock()
fn = Mock()
- class User(object):
+ class User:
pass
event.listen(User, "load", fn)
def my_listener_one(*arg, **kw):
pass
- class NotASession(object):
+ class NotASession:
def __call__(self):
return fixture_session()
assert not insp.is_aliased_class
def test_mapper_selectable_fixed(self):
- class Foo(object):
+ class Foo:
pass
class Bar(Foo):
assert insp.is_aliased_class
def test_not_mapped_class(self):
- class Foo(object):
+ class Foo:
pass
assert_raises_message(
)
def test_not_mapped_instance(self):
- class Foo(object):
+ class Foo:
pass
assert_raises_message(
class Thing(InspectionAttr):
pass
- class AnonClass(object):
+ class AnonClass:
__foo__ = "bar"
__bat__ = Thing()
)
def test_all_orm_descriptors_pep520_classical(self):
- class MyClass(object):
+ class MyClass:
pass
from sqlalchemy import Table, MetaData, Column, Integer
def test_ai(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_A(self):
inits = []
- class A(object):
+ class A:
pass
self.register(A, inits)
def test_Ai(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_ai_B(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_ai_Bi(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_bi(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_Bi(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_B(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_Bi_Ci(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_bi_Ci(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_b_Ci(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_B_Ci(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_Ai_B_C(self):
inits = []
- class A(object):
+ class A:
def __init__(self):
inits.append((A, "__init__"))
def test_A_Bi_C(self):
inits = []
- class A(object):
+ class A:
pass
self.register(A, inits)
def test_A_B_Ci(self):
inits = []
- class A(object):
+ class A:
pass
self.register(A, inits)
def test_A_B_C(self):
inits = []
- class A(object):
+ class A:
pass
self.register(A, inits)
eq_(inits, [(C, "init", C)])
def test_defaulted_init(self):
- class X(object):
+ class X:
def __init__(self_, a, b=123, c="abc"):
self_.a = a
self_.b = b
eq_(o.b, 123)
eq_(o.c, "abc")
- class Y(object):
+ class Y:
unique = object()
- class OutOfScopeForEval(object):
+ class OutOfScopeForEval:
def __repr__(self_):
# misleading repr
return "123"
)
def test_partially_mapped_inheritance(self):
- class A(object):
+ class A:
pass
class B(A):
assert_raises(sa.orm.exc.UnmappedClassError, class_mapper, C)
def test_del_warning(self):
- class A(object):
+ class A:
def __del__(self):
pass
global A
- class A(object):
+ class A:
pass
def canary(instance):
class NativeInstrumentationTest(fixtures.MappedTest):
def test_register_reserved_attribute(self):
- class T(object):
+ class T:
pass
instrumentation.register_class(T)
Column(instrumentation.ClassManager.STATE_ATTR, Integer),
)
- class T(object):
+ class T:
pass
assert_raises(KeyError, self.mapper_registry.map_imperatively, T, t)
Column(instrumentation.ClassManager.MANAGER_ATTR, Integer),
)
- class T(object):
+ class T:
pass
assert_raises(KeyError, self.mapper_registry.map_imperatively, T, t)
exec(
"""
def _kw_only_fixture(self):
- class A(object):
+ class A:
def __init__(self, a, *, b, c):
self.a = a
self.b = b
return self._instrument(A)
def _kw_plus_posn_fixture(self):
- class A(object):
+ class A:
def __init__(self, a, *args, b, c):
self.a = a
self.b = b
return self._instrument(A)
def _kw_opt_fixture(self):
- class A(object):
+ class A:
def __init__(self, a, *, b, c="c"):
self.a = a
self.b = b
Column("x", Integer),
)
- class A(object):
+ class A:
pass
self.mapper_registry.map_imperatively(A, t)
Column("t1_id", Integer, ForeignKey("t1.id")),
)
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
self.mapper_registry.map_imperatively(
assert not a.bs
def test_uninstrument(self):
- class A(object):
+ class A:
pass
manager = instrumentation.register_class(A)
Column("t1_id", Integer, ForeignKey("t1.id")),
)
- class Base(object):
+ class Base:
def __init__(self, *args, **kwargs):
pass
Column("t1_id", Integer, ForeignKey("t1.id")),
)
- class Base(object):
+ class Base:
def __init__(self):
pass
- class Base_AKW(object):
+ class Base_AKW:
def __init__(self, *args, **kwargs):
pass
Column("id", Integer, ForeignKey("base.id"), primary_key=True),
)
- class Base(object):
+ class Base:
pass
class A(Base):
if use_indirect_access:
def query(names):
- class Foo(object):
+ class Foo:
def __init__(self):
self.u1 = aliased(User)
"""
@registry.mapped
- class A(object):
+ class A:
__tablename__ = "a"
id = Column(Integer, primary_key=True)
b = relationship("B")
@registry.mapped
- class B(object):
+ class B:
__tablename__ = "b"
id = Column(Integer, primary_key=True)
from sqlalchemy.orm.base import _is_mapped_class, _is_aliased_class
- class Foo(object):
+ class Foo:
x = "something"
@property
from sqlalchemy.orm.base import _entity_descriptor
- class Foo(object):
+ class Foo:
x = "something"
@property
users, addresses = self.tables.users, self.tables.addresses
- class Foo(object):
+ class Foo:
def __init__(self):
pass
- class Bar(object):
+ class Bar:
pass
self.mapper(Foo, users)
"""
- class Foo(object):
+ class Foo:
def __init__(self, id_):
self.id = id_
def test_class_hier_only_instrument_once_multiple_configure(self):
users, addresses = (self.tables.users, self.tables.addresses)
- class A(object):
+ class A:
pass
class ASub(A):
class ASubSub(ASub):
pass
- class B(object):
+ class B:
pass
from sqlalchemy.testing import mock
users = self.tables.users
Address = self.classes.Address
- class MyComposite(object):
+ class MyComposite:
pass
for constructor, args in [
assert_calls = []
- class Address(object):
+ class Address:
def _get_user(self):
assert_calls.append("get")
return self._user
Column("parent_id", Integer, ForeignKey("nodes.id")),
)
- class Node(object):
+ class Node:
pass
self.mapper(
Column("vendor_id", Integer),
)
- class Person(object):
+ class Person:
pass
class Vendor(Person):
class Manager(Employee):
pass
- class Hoho(object):
+ class Hoho:
pass
- class Lala(object):
+ class Lala:
pass
- class Fub(object):
+ class Fub:
pass
- class Frob(object):
+ class Frob:
pass
- class HasDef(object):
+ class HasDef:
def name(self):
pass
- class Empty(object):
+ class Empty:
pass
self.mapper(
t.create(connection)
- class A(object):
+ class A:
pass
self.mapper(A, t, include_properties=["id"])
s.commit()
def test_we_dont_call_bool(self):
- class NoBoolAllowed(object):
+ class NoBoolAllowed:
def __bool__(self):
raise Exception("nope")
assert s.get(NoBoolAllowed, u1.id) is u1
def test_we_dont_call_eq(self):
- class NoEqAllowed(object):
+ class NoEqAllowed:
def __eq__(self, other):
raise Exception("nope")
class extendedproperty(property):
attribute = 123
- class User(object):
+ class User:
def _get_name(self):
assert_col.append(("get", self.name))
return self.name
# synonym points to non-existent attrbute that hasn't been mapped yet.
users = self.tables.users
- class User(object):
+ class User:
def _x(self):
return self.id
assert hasattr(User.x, "comparator")
def test_synonym_of_non_property_raises(self):
- class User(object):
+ class User:
@property
def x(self):
return "hi"
assert_col = []
- class User(object):
+ class User:
def _get_name(self):
assert_col.append(("get", self._name))
return self._name
recon = []
- class User(object):
+ class User:
@reconstructor
def reconstruct(self):
recon.append("go")
recon = []
- class A(object):
+ class A:
@reconstructor
def reconstruct(self):
assert isinstance(self, A)
recon = []
- class User(object):
+ class User:
@reconstructor
def __init__(self):
recon.append("go")
recon = []
- class recon_obj(object):
+ class recon_obj:
def __call__(self, obj):
recon.append("go")
- class User(object):
+ class User:
__init__ = reconstructor(recon_obj())
self.mapper(User, users)
recon = []
- class recon_obj(object):
+ class recon_obj:
def __call__(self, obj):
recon.append("go")
__sa_reconstructor__ = True
- class recon_meth(object):
+ class recon_meth:
__func__ = recon_obj()
def __call__(self, *arg, **kw):
return self.__func__.__call__(*arg, **kw)
- class User(object):
+ class User:
__init__ = recon_meth()
self.mapper(User, users)
recon = []
- class A(object):
+ class A:
@reconstructor
def __init__(self):
assert isinstance(self, A)
recon = []
- class Base(object):
+ class Base:
@reconstructor
def reconstruct(self):
recon.append("go")
def test_unmapped_subclass_error_postmap(self):
users = self.tables.users
- class Base(object):
+ class Base:
pass
class Sub(Base):
def test_unmapped_subclass_error_premap(self):
users = self.tables.users
- class Base(object):
+ class Base:
pass
self.mapper(Base, users)
class OldStyle:
pass
- class NewStyle(object):
+ class NewStyle:
pass
class A(NewStyle, OldStyle):
Column("value", String(10)),
)
- class _ValueBase(object):
+ class _ValueBase:
def __init__(self, value="abc", id_=None):
self.id = id_
self.value = value
def test_nonzero_len_recursion(self):
ht1 = self.tables.ht1
- class H1(object):
+ class H1:
def __len__(self):
return len(self.get_value())
self.value = "foobar"
return self.value
- class H2(object):
+ class H2:
def __bool__(self):
return bool(self.get_value())
)
def _test(self, value, instancelevel=None):
- class Foo(object):
+ class Foo:
someprop = value
m = self.mapper(Foo, self.tables.foo)
assert self.tables.foo.c.someprop not in m._columntoproperty
def _test_not(self, value):
- class Foo(object):
+ class Foo:
someprop = value
m = self.mapper(Foo, self.tables.foo)
Column(reserved, Integer),
)
- class T(object):
+ class T:
pass
clear_mappers()
sa.orm.instrumentation.ClassManager.MANAGER_ATTR,
):
- class M(object):
+ class M:
pass
clear_mappers()
),
)
- class Foo(object):
+ class Foo:
pass
- class Bar(object):
+ class Bar:
pass
self.mapper(
def test_kwarg_accepted(self):
users, Address = self.tables.users, self.classes.Address
- class DummyComposite(object):
+ class DummyComposite:
def __init__(self, x, y):
pass
ab = bc = True
@reg1.mapped
- class A(object):
+ class A:
__tablename__ = "a"
id = Column(Integer, primary_key=True)
@reg2.mapped
- class B(object):
+ class B:
__tablename__ = "b"
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey(A.id))
@reg3.mapped
- class C(object):
+ class C:
__tablename__ = "c"
id = Column(Integer, primary_key=True)
b_id = Column(ForeignKey(B.id))
def test_synonym(self):
users = self.tables.users
- class User(object):
+ class User:
def _getValue(self):
return self._value
from .inheritance._poly_fixtures import Person
-class _PolymorphicTestBase(object):
+class _PolymorphicTestBase:
__dialect__ = "default"
def test_any_one(self):
)
-class PathTest(object):
+class PathTest:
def _make_path(self, path):
r = []
for i, item in enumerate(path):
pickle.loads(pickle.dumps(screen2))
def test_exceptions(self):
- class Foo(object):
+ class Foo:
pass
users = self.tables.users
@testing.fixture
def uname_fixture(self):
- class Foo(object):
+ class Foo:
pass
if False:
def test_unique_param_names(self):
users = self.tables.users
- class SomeUser(object):
+ class SomeUser:
pass
s = users.select().where(users.c.id != 12).alias("users")
return MyQuery
def _callable_fixture(self):
- class MyQueryFactory(object):
+ class MyQueryFactory:
def __call__(self, *arg, **kw):
return Query(*arg, **kw)
from sqlalchemy.testing import mock
-class _JoinFixtures(object):
+class _JoinFixtures:
@classmethod
def setup_test_class(cls):
m = MetaData()
def mixin_fixture(self):
users = self.tables.users
- class HasFoob(object):
+ class HasFoob:
name = Column(String)
class UserWFoob(HasFoob, self.Comparable):
def declattr_mixin_fixture(self):
users = self.tables.users
- class HasFoob(object):
+ class HasFoob:
@declared_attr
def name(cls):
return Column(String)
orders, items = self.tables.orders, self.tables.items
order_items = self.tables.order_items
- class HasFoob(object):
+ class HasFoob:
description = Column(String)
class HasBat(HasFoob):
class TemporalFixtureTest(testing.fixtures.DeclarativeMappedTest):
@classmethod
def setup_classes(cls):
- class HasTemporal(object):
+ class HasTemporal:
"""Mixin that identifies a class as having a timestamp column"""
timestamp = Column(
from test.orm import _fixtures
-class _RelationshipErrors(object):
+class _RelationshipErrors:
def _assert_raises_no_relevant_fks(
self, fn, expr, relname, primary, *arg, **kw
):
)
t2 = Table("t2", m, Column("id", Integer, primary_key=True))
- class C1(object):
+ class C1:
pass
- class C2(object):
+ class C2:
pass
registry.map_imperatively(
Column("t1id", Integer, ForeignKey("t1.id")),
)
- class C1(object):
+ class C1:
pass
- class C2(object):
+ class C2:
pass
registry.map_imperatively(C1, t1, properties={"c2": relationship(C2)})
Column("t1id", Integer),
)
- class C1(object):
+ class C1:
pass
- class C2(object):
+ class C2:
pass
registry.map_imperatively(C1, t1, properties={"c2": relationship(C2)})
Column("t2_id", Integer, ForeignKey("t2.id")),
)
- class A(object):
+ class A:
pass
- class B(object):
+ class B:
pass
registry.map_imperatively(B, t2)
def test_composite_property_flag(self):
Order, orders = self.classes.Order, self.tables.orders
- class MyComposite(object):
+ class MyComposite:
def __init__(self, description, isopen):
self.description = description
self.isopen = isopen
sess = fixture_session()
- class Foo(object):
+ class Foo:
def __init__(self):
sess.add(self)
# flush will no-op without something in the unit of work
def _():
- class OK(object):
+ class OK:
pass
self._map_it(OK)
)
def test_unmapped_instance(self):
- class Unmapped(object):
+ class Unmapped:
pass
self._test_instance_guards(Unmapped())
self._test_class_guards(prim, is_class=False)
def test_unmapped_class_for_instance(self):
- class Unmapped(object):
+ class Unmapped:
pass
self._test_instance_guards(Unmapped)
self._test_class_guards(Unmapped)
def test_mapped_class_for_instance(self):
- class Mapped(object):
+ class Mapped:
pass
self._map_it(Mapped)
# no class guards- it would pass.
def test_missing_state(self):
- class Mapped(object):
+ class Mapped:
pass
early = Mapped()
self._test_class_guards(early, is_class=False)
def test_refresh_arg_signature(self):
- class Mapped(object):
+ class Mapped:
pass
self._map_it(Mapped)
def test_basic(self):
t2, t1 = self.tables.t2, self.tables.t1
- class T(object):
+ class T:
pass
class SubT(T):
pass
- class T2(object):
+ class T2:
pass
class SubT2(T2):
from sqlalchemy.testing.schema import Table
-class AssertsUOW(object):
+class AssertsUOW:
def _get_test_uow(self, session):
uow = unitofwork.UOWTransaction(session)
deleted = set(session._deleted)
assert s.identity_map[identity_key(User, ("u1",))] is u1
-class JoinIntoAnExternalTransactionFixture(object):
+class JoinIntoAnExternalTransactionFixture:
"""Test the "join into an external transaction" examples"""
__leave_connections_for_teardown__ = True
# begin a non-ORM transaction
self.trans = self.connection.begin()
- class A(object):
+ class A:
pass
clear_mappers()
# begin a non-ORM transaction
self.trans = self.connection.begin()
- class A(object):
+ class A:
pass
# TODO: py2 is not hitting this correctly for some reason,
from test.orm import _fixtures
-class UnitOfWorkTest(object):
+class UnitOfWorkTest:
pass
eq_(hb.value, False)
def test_clauseelement_accessor(self):
- class Thing(object):
+ class Thing:
def __init__(self, value):
self.value = value
names = []
- class Events(object):
+ class Events:
def before_insert(self, mapper, connection, instance):
self.current_instance = instance
names.append(instance.name)
class EnsurePKSortableTest(fixtures.MappedTest):
- class SomeEnum(object):
+ class SomeEnum:
# Implements PEP 435 in the minimal fashion needed by SQLAlchemy
__members__ = OrderedDict()
from test.orm import _fixtures
-class AssertsUOW(object):
+class AssertsUOW:
def _get_test_uow(self, session):
uow = unitofwork.UOWTransaction(session)
deleted = set(session._deleted)
def define_tables(cls, metadata):
from sqlalchemy import TypeDecorator
- class NoBool(object):
+ class NoBool:
def __nonzero__(self):
raise NotImplementedError("not supported")
- class MyWidget(object):
+ class MyWidget:
def __init__(self, text):
self.text = text
def test_evaluate_clauseelement(self):
User = self.classes.User
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return User.name.__clause_element__()
def test_evaluate_invalid(self):
User = self.classes.User
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return 5
eq_(jill.name, "moonbeam")
def test_evaluate_synonym_string(self):
- class Foo(object):
+ class Foo:
pass
self.mapper_registry.map_imperatively(
eq_(jill.uname, "moonbeam")
def test_evaluate_synonym_attr(self):
- class Foo(object):
+ class Foo:
pass
self.mapper_registry.map_imperatively(
eq_(jill.uname, "moonbeam")
def test_evaluate_double_synonym_attr(self):
- class Foo(object):
+ class Foo:
pass
self.mapper_registry.map_imperatively(
return table
def test_simple(self):
- class Point(object):
+ class Point:
pass
table = self._fixture(Point)
assert alias.id.__clause_element__().table is not table
def test_named_entity(self):
- class Point(object):
+ class Point:
pass
self._fixture(Point)
)
def test_named_selectable(self):
- class Point(object):
+ class Point:
pass
table = self._fixture(Point)
)
def test_not_instantiatable(self):
- class Point(object):
+ class Point:
pass
self._fixture(Point)
assert_raises(TypeError, alias)
def test_instancemethod(self):
- class Point(object):
+ class Point:
def zero(self):
self.x, self.y = 0, 0
assert getattr(alias, "zero")
def test_classmethod(self):
- class Point(object):
+ class Point:
@classmethod
def max_x(cls):
return 100
assert Point.max_x() == alias.max_x() == 100
def test_simple_property(self):
- class Point(object):
+ class Point:
@property
def max_x(self):
return 100
assert Point.max_x is alias.max_x
def test_descriptors(self):
- class descriptor(object):
+ class descriptor:
def __init__(self, fn):
self.fn = fn
def method(self):
return "method"
- class Point(object):
+ class Point:
center = (0, 0)
@descriptor
assert child.table is table
def test_hybrid_descriptor_one(self):
- class Point(object):
+ class Point:
def __init__(self, x, y):
self.x, self.y = x, y
)
def test_hybrid_descriptor_two(self):
- class Point(object):
+ class Point:
def __init__(self, x, y):
self.x, self.y = x, y
)
def test_hybrid_descriptor_three(self):
- class Point(object):
+ class Point:
def __init__(self, x, y):
self.x, self.y = x, y
)
def test_proxy_descriptor_one(self):
- class Point(object):
+ class Point:
def __init__(self, x, y):
self.x, self.y = x, y
)
def test_parententity_vs_parentmapper(self):
- class Point(object):
+ class Point:
pass
self._fixture(Point, properties={"x_syn": synonym("x")})
users = self.tables.users
canary = Mock()
- class SomeValidator(object):
+ class SomeValidator:
def __call__(self, obj, key, name):
canary(key, name)
ne_(name, "fred")
def test_validator_multi_warning(self):
users = self.tables.users
- class Foo(object):
+ class Foo:
@validates("name")
def validate_one(self, key, value):
pass
users,
)
- class Bar(object):
+ class Bar:
@validates("id")
def validate_three(self, key, value):
return value + 10
self.arg = arg
-class CoreFixtures(object):
+class CoreFixtures:
# lambdas which return a tuple of ColumnElement objects.
# must return at least two objects that should compare differently.
# to test more varieties of "difference" additional objects can be added.
]
-class CacheKeyFixture(object):
+class CacheKeyFixture:
def _compare_equal(self, a, b, compare_values):
a_key = a._generate_cache_key()
b_key = b._generate_cache_key()
def fn2(x, y, z=3):
pass
- class fn3(object):
+ class fn3:
def __init__(self, x, y):
pass
- class FN4(object):
+ class FN4:
def __call__(self, x, y):
pass
fn5 = list
- class fn6a(object):
+ class fn6a:
def __init__(self, x):
eq_(x, "context")
- class fn6b(object):
+ class fn6b:
def __init__(self, x, y=3):
eq_(x, "context")
- class FN7(object):
+ class FN7:
def __call__(self, x):
eq_(x, "context")
fn7 = FN7()
- class FN8(object):
+ class FN8:
def __call__(self, x, y=3):
eq_(x, "context")
use_function_defaults = testing.against("postgresql", "mssql")
is_oracle = testing.against("oracle")
- class MyClass(object):
+ class MyClass:
@classmethod
def gen_default(cls, ctx):
return "hi"
from sqlalchemy.testing.schema import Table
-class _DeleteTestBase(object):
+class _DeleteTestBase:
@classmethod
def define_tables(cls, metadata):
Table(
def test_new_calling_style_thing_ok_actually_use_iter(self, table_fixture):
table1, table2 = table_fixture
- class Thing(object):
+ class Thing:
def __iter__(self):
return iter([table1.c.name, table1.c.description])
from sqlalchemy.testing import fixtures
-class ORMExpr(object):
+class ORMExpr:
def __init__(self, col):
self.col = col
return self.col
-class _InsertTestBase(object):
+class _InsertTestBase:
@classmethod
def define_tables(cls, metadata):
Table(
# refer to unknown types of objects inside the lambda. these have
# to be resolved outside of the lambda because we otherwise can't
# be sure they can be safely used as cache keys.
- class Thing(object):
+ class Thing:
def __init__(self, col_expr):
self.col_expr = col_expr
# refer to unknown types of objects inside the lambda. these have
# to be resolved outside of the lambda because we otherwise can't
# be sure they can be safely used as cache keys.
- class Thing(object):
+ class Thing:
def __init__(self, col_expr):
self.col_expr = col_expr
# test the above 'test_reject_plain_object' with the expected
# workaround
- class Thing(object):
+ class Thing:
def __init__(self, col_expr):
self.col_expr = col_expr
def test_cache_key_instance_variable_issue_incorrect(self):
t1 = table("t1", column("q"), column("p"))
- class Foo(object):
+ class Foo:
def __init__(self, value):
self.value = value
def test_cache_key_instance_variable_issue_correct_one(self):
t1 = table("t1", column("q"), column("p"))
- class Foo(object):
+ class Foo:
def __init__(self, value):
self.value = value
def test_cache_key_instance_variable_issue_correct_two(self):
t1 = table("t1", column("q"), column("p"))
- class Foo(object):
+ class Foo:
def __init__(self, value):
self.value = value
)
def test_fk_given_non_col_clauseelem(self):
- class Foo(object):
+ class Foo:
def __clause_element__(self):
return bindparam("x")
def test_fk_given_col_non_table_clauseelem(self):
t = Table("t", MetaData(), Column("x", Integer))
- class Foo(object):
+ class Foo:
def __clause_element__(self):
return t.alias().c.x
class SchemaTypeTest(fixtures.TestBase):
__backend__ = True
- class TrackEvents(object):
+ class TrackEvents:
column = None
table = None
evt_targets = ()
def test_clauseelement_extraction_one(self):
t = Table("t", MetaData(), Column("x", Integer), Column("y", Integer))
- class MyThing(object):
+ class MyThing:
def __clause_element__(self):
return t.c.x + 5
def test_clauseelement_extraction_two(self):
t = Table("t", MetaData(), Column("x", Integer), Column("y", Integer))
- class MyThing(object):
+ class MyThing:
def __clause_element__(self):
return t.c.x + 5
expr1 = t.c.x + 5
- class MyThing(object):
+ class MyThing:
def __clause_element__(self):
return expr1
def test_column_accessor_clause_element(self):
c1 = Column("x", Integer)
- class CThing(object):
+ class CThing:
def __init__(self, c):
self.c = c
m = MetaData()
t2 = Table("t2", m, Column("x", Integer))
- class SomeClass(object):
+ class SomeClass:
def __clause_element__(self):
return t2
@testing.fixture
def no_pickle_annotated(self):
- class NoPickle(object):
+ class NoPickle:
def __reduce__(self):
raise NotImplementedError()
)
-class _CustomComparatorTests(object):
+class _CustomComparatorTests:
def test_override_builtin(self):
c1 = Column("foo", self._add_override_factory())
self._assert_add_override(c1)
)
def test_in_arbitrary_sequence(self):
- class MySeq(object):
+ class MySeq:
def __init__(self, d):
self.d = d
result = connection.execute(select(1))
row = result.first()
- class unprintable(object):
+ class unprintable:
def __str__(self):
raise ValueError("nope")
"""
- class MyList(object):
+ class MyList:
def __init__(self, data):
self.internal_list = data
argnames="method_name",
)
def test_handle_error_in_fetch(self, strategy_cls, method_name):
- class cursor(object):
+ class cursor:
def raise_(self):
raise IOError("random non-DBAPI error during cursor operation")
t = Table("t", m, Column("q", Integer))
-class NotAThing1(object):
+class NotAThing1:
pass
not_a_thing2 = NotAThing2()
-class NotAThing3(object):
+class NotAThing3:
def __clause_element__(self):
return not_a_thing2
def some_function():
pass
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return some_function
)
def test_new_calling_style_clauseelement_thing_that_has_iter(self):
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return table1
)
def test_new_calling_style_inspectable_ce_thing_that_has_iter(self):
- class Thing(object):
+ class Thing:
def __iter__(self):
return iter(["a", "b", "c"])
- class InspectedThing(object):
+ class InspectedThing:
def __clause_element__(self):
return table1
from sqlalchemy.testing import fixtures
-class _ExprFixture(object):
+class _ExprFixture:
def _test_table(self, type_):
test_table = Table(
"test_table", MetaData(), Column("x", String), Column("y", type_)
)
-class RoundTripTestBase(object):
+class RoundTripTestBase:
def test_round_trip(self, connection):
connection.execute(
self.tables.test_table.insert(),
loads(dumps(meta))
-class _UserDefinedTypeFixture(object):
+class _UserDefinedTypeFixture:
@classmethod
def define_tables(cls, metadata):
class MyType(types.UserDefinedType):
# test coerce from nulltype - e.g. use an object that
# doesn't match to a known type
- class MyObj(object):
+ class MyObj:
def __str__(self):
return "THISISMYOBJ"
t = self.tables.t
conn.execute(t.insert().values(data=coerce_fn("d1", MyType)))
- class MyFoob(object):
+ class MyFoob:
def __clause_element__(self):
return t.c.data
m1.create_all(testing.db)
def test_non_native_constraint_custom_type(self):
- class Foob(object):
+ class Foob:
def __init__(self, name):
self.name = name
class MyFoobarType(types.UserDefinedType):
pass
- class Foo(object):
+ class Foo:
pass
# unknown type + integer, right hand bind
def test_detect_coercion_of_builtins(self):
@inspection._self_inspects
- class SomeSQLAThing(object):
+ class SomeSQLAThing:
def __repr__(self):
return "some_sqla_thing()"
- class SomeOtherThing(object):
+ class SomeOtherThing:
pass
assert_raises_message(
)
def test_non_native_constraint_custom_type(self):
- class Foob(object):
+ class Foob:
def __init__(self, value):
self.value = value
from sqlalchemy.testing.schema import Table
-class _UpdateFromTestBase(object):
+class _UpdateFromTestBase:
@classmethod
def define_tables(cls, metadata):
Table(
def test_update_custom_key_thing(self):
table1 = self.tables.mytable
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return table1.c.name
def test_update_ordered_custom_key_thing(self):
table1 = self.tables.mytable
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return table1.c.name
def test_update_broken_custom_key_thing(self):
table1 = self.tables.mytable
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return 5
def test_update_ordered_broken_custom_key_thing(self):
table1 = self.tables.mytable
- class Thing(object):
+ class Thing:
def __clause_element__(self):
return 5
@testing.fixture
def tricky_types_parameter_fixture(self):
- class SomeEnum(object):
+ class SomeEnum:
# Implements PEP 435 in the minimal fashion needed by SQLAlchemy
__members__ = OrderedDict()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
