本文整理汇总了Python中sqlalchemy.ext.baked.bakery函数的典型用法代码示例。如果您正苦于以下问题:Python bakery函数的具体用法?Python bakery怎么用?Python bakery使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了bakery函数的14个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_subqueryload_post_context
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)
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:29,代码来源:test_baked.py
示例2: test_baked_query
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()
开发者ID:Dorthu,项目名称:sqlalchemy,代码行数:8,代码来源:short_selects.py
示例3: test_baked_query_cols_only
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()
开发者ID:Dorthu,项目名称:sqlalchemy,代码行数:10,代码来源:short_selects.py
示例4: setup
def setup(self):
self.bakery = baked.bakery()
开发者ID:CyberCollins,项目名称:sqlalchemy,代码行数:2,代码来源:test_baked.py
示例5: RDBStorage
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
开发者ID:ENCODE-DCC,项目名称:snovault,代码行数:31,代码来源:storage.py
示例6: import
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.
开发者ID:openstack,项目名称:group-based-policy,代码行数:32,代码来源:aim_mapping_rpc.py
示例7: import
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.,
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:31,代码来源:grouper.py
示例8: getbakery
def getbakery(cls):
if cls._bakery is None:
cls._bakery = staticmethod(bakery())
return cls._bakery
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:4,代码来源:generic.py
示例9: import
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)
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:29,代码来源:glycomics.py
示例10: test_subquery_eagerloading
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()
开发者ID:robin900,项目名称:sqlalchemy,代码行数:74,代码来源:test_baked.py
示例11: User
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)
开发者ID:mikegreen7892003,项目名称:PythonToy,代码行数:30,代码来源:sqlalchemy_bakery.py
示例12: import
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 = []
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:31,代码来源:observed_ions.py
示例13: import
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):
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:31,代码来源:peptide.py
示例14: bakery
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):
开发者ID:BostonUniversityCBMS,项目名称:glycresoft_sqlalchemy,代码行数:31,代码来源:sequencing.py
注:本文中的sqlalchemy.ext.baked.bakery函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论