Skip to content

Latest commit

 

History

History
222 lines (165 loc) · 8.02 KB

File metadata and controls

222 lines (165 loc) · 8.02 KB

title: sqlalchemy.orm class_mapper Example Code category: page slug: sqlalchemy-orm-class-mapper-examples sortorder: 500031064 toc: False sidebartitle: sqlalchemy.orm class_mapper meta: Python example code that shows how to use the class_mapper callable from the sqlalchemy.orm module of the SQLAlchemy project.

class_mapper is a callable within the sqlalchemy.orm module of the SQLAlchemy project.

ColumnProperty, CompositeProperty, Load, Mapper, Query, RelationshipProperty, Session, SynonymProperty, aliased, attributes, backref, column_property, composite, interfaces, mapper, mapperlib, object_mapper, object_session, query, relationship, session, sessionmaker, and strategies are several other callables with code examples from the same sqlalchemy.orm package.

Example 1 from graphene-sqlalchemy

graphene-sqlalchemy (project documentation and PyPI package information) is a SQLAlchemy integration for Graphene, which makes it easier to build GraphQL-based APIs into Python web applications. The package allows you to subclass SQLAlchemy classes and build queries around them with custom code to match the backend queries with the GraphQL-based request queries. The project is provided as open source under the MIT license.

graphene-sqlalchemy / graphene_sqlalchemy / utils.py

# utils.py
import re
import warnings

from sqlalchemy.exc import ArgumentError
~~from sqlalchemy.orm import class_mapper, object_mapper
from sqlalchemy.orm.exc import UnmappedClassError, UnmappedInstanceError


def get_session(context):
    return context.get("session")


def get_query(model, context):
    query = getattr(model, "query", None)
    if not query:
        session = get_session(context)
        if not session:
            raise Exception(
                "A query in the model Base or a session in the schema is required for querying.\n"
                "Read more http://docs.graphene-python.org/projects/sqlalchemy/en/latest/tips/#querying"
            )
        query = session.query(model)
    return query


def is_mapped_class(cls):
    try:
~~        class_mapper(cls)
    except (ArgumentError, UnmappedClassError):
        return False
    else:
        return True


def is_mapped_instance(cls):
    try:
        object_mapper(cls)
    except (ArgumentError, UnmappedInstanceError):
        return False
    else:
        return True


def to_type_name(name):
    return "".join(part[:1].upper() + part[1:] for part in name.split("_"))


_re_enum_value_name_1 = re.compile("(.)([A-Z][a-z]+)")
_re_enum_value_name_2 = re.compile("([a-z0-9])([A-Z])")


def to_enum_value_name(name):


## ... source file continues with no further class_mapper examples...

Example 2 from sqlalchemy-utils

sqlalchemy-utils (project documentation and PyPI package information) is a code library with various helper functions and new data types that make it easier to use SQLAlchemy when building projects that involve more specific storage requirements such as currency. The wide array of data types includes ranged values and aggregated attributes.

sqlalchemy-utils / sqlalchemy_utils / generic.py

# generic.py
from collections.abc import Iterable

import six
import sqlalchemy as sa
from sqlalchemy.ext.hybrid import hybrid_property
~~from sqlalchemy.orm import attributes, class_mapper, ColumnProperty
from sqlalchemy.orm.interfaces import MapperProperty, PropComparator
from sqlalchemy.orm.session import _state_session
from sqlalchemy.util import set_creation_order

from .exceptions import ImproperlyConfigured
from .functions import identity


class GenericAttributeImpl(attributes.ScalarAttributeImpl):
    def get(self, state, dict_, passive=attributes.PASSIVE_OFF):
        if self.key in dict_:
            return dict_[self.key]

        session = _state_session(state)
        if session is None:
            return None

        discriminator = self.get_state_discriminator(state)
        target_class = state.class_._decl_class_registry.get(discriminator)

        if target_class is None:
            return None

        id = self.get_state_id(state)


## ... source file abbreviated to get to class_mapper examples ...


        return target

    def get_state_discriminator(self, state):
        discriminator = self.parent_token.discriminator
        if isinstance(discriminator, hybrid_property):
            return getattr(state.obj(), discriminator.__name__)
        else:
            return state.attrs[discriminator.key].value

    def get_state_id(self, state):
        return tuple(state.attrs[id.key].value for id in self.parent_token.id)

    def set(self, state, dict_, initiator,
            passive=attributes.PASSIVE_OFF,
            check_old=None,
            pop=False):

        dict_[self.key] = initiator

        if initiator is None:
            for id in self.parent_token.id:
                dict_[id.key] = None
            dict_[self.parent_token.discriminator.key] = None
        else:
            class_ = type(initiator)
~~            mapper = class_mapper(class_)

            pk = mapper.identity_key_from_instance(initiator)[1]

            discriminator = six.text_type(class_.__name__)

            for index, id in enumerate(self.parent_token.id):
                dict_[id.key] = pk[index]
            dict_[self.parent_token.discriminator.key] = discriminator


class GenericRelationshipProperty(MapperProperty):

    def __init__(self, discriminator, id, doc=None):
        super(GenericRelationshipProperty, self).__init__()
        self._discriminator_col = discriminator
        self._id_cols = id
        self._id = None
        self._discriminator = None
        self.doc = doc

        set_creation_order(self)

    def _column_to_property(self, column):
        if isinstance(column, hybrid_property):


## ... source file continues with no further class_mapper examples...