def test_subqueryload_post_context(self):
        User = self.classes.User
        Address = self.classes.Address

        assert_result = [
            User(
                id=7, addresses=[Address(id=1, email_address="[email protected]")]
            )
        ]

        self.bakery = baked.bakery(size=3)

        bq = self.bakery(lambda s: s.query(User))

        bq += lambda q: q.options(subqueryload(User.addresses))
        bq += lambda q: q.order_by(User.id)
        bq += lambda q: q.filter(User.name == bindparam("name"))
        sess = Session()

        def set_params(q):
            return q.params(name="jack")

        # test that the changes we make using with_post_criteria()
        # are also applied to the subqueryload query.
        def go():
            result = bq(sess).with_post_criteria(set_params).all()
            eq_(assert_result, result)

        self.assert_sql_count(testing.db, go, 2)

def test_baked_query(n):
    """test a baked query of the full entity."""
    bakery = baked.bakery()
    s = Session(bind=engine)
    for id_ in random.sample(ids, n):
        q = bakery(lambda s: s.query(Customer))
        q += lambda q: q.filter(Customer.id == bindparam('id'))
        q(s).params(id=id_).one()

def test_baked_query_cols_only(n):
    """test a baked query of only the entity columns."""
    bakery = baked.bakery()
    s = Session(bind=engine)
    for id_ in random.sample(ids, n):
        q = bakery(
            lambda s: s.query(
                Customer.id, Customer.name, Customer.description))
        q += lambda q: q.filter(Customer.id == bindparam('id'))
        q(s).params(id=id_).one()

 def setup(self):
     self.bakery = baked.bakery()

    registry[STORAGE] = RDBStorage(registry[DBSESSION])
    global _DBSESSION
    _DBSESSION = registry[DBSESSION]
    if registry.settings.get('blob_bucket'):
        registry[BLOBS] = S3BlobStorage(
            registry.settings['blob_bucket'],
            read_profile_name=registry.settings.get('blob_read_profile_name'),
            store_profile_name=registry.settings.get('blob_store_profile_name')
        )
    else:
        registry[BLOBS] = RDBBlobStorage(registry[DBSESSION])


Base = declarative_base()

bakery = baked.bakery()
baked_query_resource = bakery(lambda session: session.query(Resource))
baked_query_unique_key = bakery(
    lambda session: session.query(Key).options(
        orm.joinedload_all(
            Key.resource,
            Resource.data,
            CurrentPropertySheet.propsheet,
            innerjoin=True,
        ),
    ).filter(Key.name == bindparam('name'), Key.value == bindparam('value'))
)


class RDBStorage(object):
    batchsize = 1000

from neutron_lib.api.definitions import portbindings
from opflexagent import rpc as o_rpc
from oslo_log import log

from gbpservice.neutron.db.grouppolicy.extensions import (
    apic_auto_ptg_db as auto_ptg_db)
from gbpservice.neutron.db.grouppolicy.extensions import (
    apic_segmentation_label_db as seg_label_db)
from gbpservice.neutron.db.grouppolicy import group_policy_mapping_db as gpmdb
from gbpservice.neutron.plugins.ml2plus.drivers.apic_aim import (
    constants as md_const)


LOG = log.getLogger(__name__)

BAKERY = baked.bakery(_size_alert=lambda c: LOG.warning(
    "sqlalchemy baked query cache size exceeded in %s" % __name__))

EndpointPtInfo = namedtuple(
    'EndpointPtInfo',
    ['pt_id',
     'ptg_id',
     'apg_id',
     'inject_default_route',
     'l3p_project_id',
     'is_auto_ptg'])


class AIMMappingRPCMixin(object):
    """RPC mixin for AIM mapping.

    Collection of all the RPC methods consumed by the AIM mapping.

from sqlalchemy import func, bindparam, select
from sqlalchemy.orm import make_transient


from glycresoft_sqlalchemy.data_model import (
    Decon2LSPeak, Decon2LSPeakGroup, Decon2LSPeakToPeakGroupMap,
    PipelineModule, MSScan, ScanBase)

from glycresoft_sqlalchemy.utils.common_math import ppm_error

from .common import (
    expanding_window, expected_a_peak_regression, centroid_scan_error_regression)

TDecon2LSPeakGroup = Decon2LSPeakGroup.__table__

query_oven = bakery()
get_group_id_by_mass_window = query_oven(lambda session: session.query(Decon2LSPeakGroup.id))
get_group_id_by_mass_window += lambda q: q.filter(Decon2LSPeakGroup.weighted_monoisotopic_mass.between(
            bindparam("lower"), bindparam("upper")))
get_group_id_by_mass_window += lambda q: q.filter(Decon2LSPeakGroup.sample_run_id == bindparam("sample_run_id"))


class Decon2LSPeakGrouper(PipelineModule):
    '''
    Pipeline Step to post-process Decon2LSPeaks, clustering them by mass
    and calculating trends across groups.
    '''
    def __init__(self, database_path, sample_run_id=1, grouping_error_tolerance=8e-5,
                 minimum_scan_count=1, max_charge_state=8,
                 minimum_abundance_ratio=0.01, minimum_mass=1200.,
                 maximum_mass=15000., minimum_signal_to_noise=1.,

 def getbakery(cls):
     if cls._bakery is None:
         cls._bakery = staticmethod(bakery())
     return cls._bakery

from .base import Base, Namespace
from .generic import (
    MutableDict, HasTaxonomy,
    HasReferenceAccessionNumber,
    HasClassBakedQueries)

from ..utils.database_utils import get_or_create
from ..utils.memoize import memoclone

FrozenGlycanComposition = glycan_composition.FrozenGlycanComposition
crossring_pattern = re.compile(r"\d,\d")
glycoct_parser = memoclone(100)(glycoct.loads)


glycan_bakery = bakery()


class MassShift(Base):
    __tablename__ = "MassShift"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(Unicode(128), index=True)
    mass = Column(Numeric(10, 6, asdecimal=False))

    def __hash__(self):
        return hash(self.name)

    def __eq__(self, other):
        return (self.name == other.name) and (self.mass == other.mass)

    def test_subquery_eagerloading(self):
        User = self.classes.User
        Address = self.classes.Address
        Order = self.classes.Order

        # Override the default bakery for one with a smaller size. This used to
        # trigger a bug when unbaking subqueries.
        self.bakery = baked.bakery(size=3)
        base_bq = self.bakery(lambda s: s.query(User))

        base_bq += lambda q: q.options(subqueryload(User.addresses),
                                       subqueryload(User.orders))
        base_bq += lambda q: q.order_by(User.id)

        assert_result = [
            User(id=7,
                addresses=[Address(id=1, email_address='[email protected]')],
                orders=[Order(id=1), Order(id=3), Order(id=5)]),
            User(id=8, addresses=[
                Address(id=2, email_address='[email protected]'),
                Address(id=3, email_address='[email protected]'),
                Address(id=4, email_address='[email protected]'),
            ]),
            User(id=9,
                addresses=[Address(id=5)],
                orders=[Order(id=2), Order(id=4)]),
            User(id=10, addresses=[])
        ]

        for i in range(4):
            for cond1, cond2 in itertools.product(
                    *[(False, True) for j in range(2)]):
                bq = base_bq._clone()

                sess = Session()

                if cond1:
                    bq += lambda q: q.filter(User.name == 'jack')
                else:
                    bq += lambda q: q.filter(User.name.like('%ed%'))

                if cond2:
                    ct = func.count(Address.id).label('count')
                    subq = sess.query(
                        ct,
                        Address.user_id).group_by(Address.user_id).\
                        having(ct > 2).subquery()

                    bq += lambda q: q.join(subq)

                if cond2:
                    if cond1:
                        def go():
                            result = bq(sess).all()
                            eq_([], result)
                        self.assert_sql_count(testing.db, go, 1)
                    else:
                        def go():
                            result = bq(sess).all()
                            eq_(assert_result[1:2], result)
                        self.assert_sql_count(testing.db, go, 3)
                else:
                    if cond1:
                        def go():
                            result = bq(sess).all()
                            eq_(assert_result[0:1], result)
                        self.assert_sql_count(testing.db, go, 3)
                    else:
                        def go():
                            result = bq(sess).all()
                            eq_(assert_result[1:3], result)
                        self.assert_sql_count(testing.db, go, 3)

                sess.close()

import logging
import pprint

import tornado.web
import tornado.httpserver
from tornado.options import define, options, parse_command_line
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext import baked
from sqlalchemy.orm import Session


# models

BAKERY = baked.bakery()


Base = declarative_base()


ENGINE = create_engine('sqlite:///:memory:', echo=False)


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)

from sqlalchemy import (PickleType, Numeric, Unicode, Table, bindparam,
                        Column, Integer, ForeignKey, UnicodeText, Boolean)

import numpy as np

from .base import Base
from .connection import DatabaseManager
from .generic import MutableDict, ParameterStore

from ..structure.composition import Composition
from ..utils.common_math import DPeak
from ..utils.collectiontools import groupby

PROTON = Composition("H+").mass

observed_ions_bakery = bakery()


class HasPeakChromatogramData(object):

    peak_data = Column(MutableDict.as_mutable(PickleType))

    def get_chromatogram(self):
        peak_data = self.peak_data
        scans = peak_data['scan_times']
        intensity = peak_data['intensities']

        scan_groups = groupby(
            zip(scans, intensity), key_fn=operator.itemgetter(0),
            transform_fn=operator.itemgetter(1))
        scans = []

from sqlalchemy.orm import relationship, backref, make_transient, Query, validates
from sqlalchemy.ext.hybrid import hybrid_method, hybrid_property
from sqlalchemy import (PickleType, Numeric, Unicode, Table, bindparam,
                        Column, Integer, ForeignKey, UnicodeText, Boolean,
                        event)
from sqlalchemy.orm.exc import DetachedInstanceError
from ..generic import MutableDict, MutableList, HasClassBakedQueries
from ..base import Base
from ..hypothesis import Hypothesis
from ..glycomics import (
    with_glycan_composition, TheoreticalGlycanCombination, has_glycan_composition_listener)

from ...structure import sequence, residue


peptide_bakery = bakery()


class Protein(Base):
    __tablename__ = "Protein"

    id = Column(Integer, primary_key=True, autoincrement=True)
    protein_sequence = Column(UnicodeText, default=u"")
    name = Column(Unicode(128), default=u"", index=True)
    other = Column(MutableDict.as_mutable(PickleType))
    hypothesis_id = Column(Integer, ForeignKey("Hypothesis.id", ondelete="CASCADE"))

    _n_glycan_sequon_sites = None

    @property
    def n_glycan_sequon_sites(self):

from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import Numeric, Unicode, Column, Integer, ForeignKey, Table, UnicodeText, bindparam

from ..base import Base
from sqlalchemy.ext.baked import bakery

from glycresoft_sqlalchemy.structure.sequence_composition import SequenceComposition, Composition


sequencing_bakery = bakery()


class SequenceBuildingBlock(Base):
    __tablename__ = "SequenceBuildingBlock"

    id = Column(Integer, primary_key=True)
    name = Column(Unicode(120), index=True)
    mass = Column(Numeric(10, 6), index=True)
    hypothesis_id = Column(Integer, ForeignKey("Hypothesis.id"), index=True)


class SequenceSegment(Base):
    __tablename__ = "SequenceSegment"
    id = Column(Integer, primary_key=True)
    sequence = Column(Unicode(128))
    mass = Column(Numeric(12, 6), index=True)
    count_n_glycosylation = Column(Integer)


class AminoAcidComposition(Base):