本文整理汇总了Python中sqlalchemy.orm.foreign函数的典型用法代码示例。如果您正苦于以下问题:Python foreign函数的具体用法?Python foreign怎么用?Python foreign使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了foreign函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: slices
def slices(self):
return relationship(
'Slice',
primaryjoin=lambda: and_(
foreign(Slice.datasource_id) == self.id,
foreign(Slice.datasource_type) == self.type,
),
)
开发者ID:bkyryliuk,项目名称:caravel,代码行数:8,代码来源:models.py
示例2: redirect
def redirect(cls):
pk = getattr(cls, cls.get_pk_name())
return relationship(
cls, foreign_keys=[cls.redirect_page],
remote_side='{}.{}'.format(cls.__name__, cls.get_pk_name()),
primaryjoin=lambda: foreign(cls.redirect_page) == pk,
)
开发者ID:uralbash,项目名称:pyramid_pages,代码行数:7,代码来源:models.py
示例3: test_dont_send_neverset_to_get_w_relationship
def test_dont_send_neverset_to_get_w_relationship(self):
# test issue #3647
CompositePk, composite_pk_table = (self.classes.CompositePk, self.tables.composite_pk_table)
User, users = (self.classes.User, self.tables.users)
mapper(
User,
users,
properties={
"elements": relationship(CompositePk, primaryjoin=users.c.id == foreign(composite_pk_table.c.i))
},
)
mapper(CompositePk, composite_pk_table)
u1 = User(id=5, name="some user")
cp1 = CompositePk(j=1, k=1)
u1.elements.append(cp1)
sess = Session()
rec = []
def go():
rec.append(sess.merge(u1))
self.assert_sql_count(testing.db, go, 1)
u2 = rec[0]
sess.commit()
eq_(u2.elements[0].i, 5)
eq_(u2.id, 5)
开发者ID:kongxianghe1234,项目名称:sqlalchemy,代码行数:28,代码来源:test_merge.py
示例4: __init__
def __init__(self, to, ct_field='model', to_field='object_id', lazy='dynamic', primary_join=None, *args, **kwargs):
if primary_join is None:
primary_join = lambda model, fk_model: and_(
model._meta.pk.column == foreign(fk_model._meta.fields[to_field].column),
fk_model._meta.fields_dict[ct_field].column == model._meta.name
)
super(GenericOneToManyField, self).__init__(to, to_field, lazy=lazy, primary_join=primary_join, *args, **kwargs)
开发者ID:katrid,项目名称:orun,代码行数:7,代码来源:fields.py
示例5: __declare_last__
def __declare_last__(cls):
from .invoice import Job
cls.jobs = relation('Job',
primaryjoin=cls.guid == foreign(Job.owner_guid),
cascade='all, delete-orphan',
collection_class=CallableList,
)
开发者ID:e2thenegpii,项目名称:piecash,代码行数:7,代码来源:person.py
示例6: setup_listener
def setup_listener(mapper, class_):
name = class_.__name__
discriminator = name.lower()
class_.addresses = relationship(Address,
primaryjoin=and_(
class_.id == foreign(remote(Address.parent_id)),
Address.discriminator == discriminator
),
backref=backref(
"parent_%s" % discriminator,
primaryjoin=remote(class_.id) == foreign(Address.parent_id)
)
)
@event.listens_for(class_.addresses, "append")
def append_address(target, value, initiator):
value.discriminator = discriminator
开发者ID:aburan28,项目名称:sqlalchemy,代码行数:16,代码来源:generic_fk.py
示例7: slots
def slots(cls):
rel = relation('Slot',
primaryjoin=foreign(Slot.obj_guid) == cls.guid,
cascade='all, delete-orphan',
collection_class=CallableList,
)
return rel
开发者ID:e2thenegpii,项目名称:piecash,代码行数:8,代码来源:_declbase.py
示例8: setup_listener
def setup_listener(mapper, class_):
name = class_.__name__
content_type = name.lower()
# Reverse relationship (ie. Segment.entities)
class_.entities = relationship(Entity,
primaryjoin = and_(
class_.id == foreign(remote(Entity.object_id)),
Entity.content_type == content_type
),
backref = backref(
'parent_{0}'.format(content_type),
primaryjoin=remote(class_.id) == foreign(Entity.object_id)
)
)
@event.listens_for(class_.entities, 'append')
def append_entity(target, value, initiator):
value.content_type = content_type
开发者ID:chrisbura,项目名称:pythoncad-legacy,代码行数:17,代码来源:schema.py
示例9: _unhashable_fixture
def _unhashable_fixture(self, metadata, load_on_pending=False):
class MyHashType(sa.TypeDecorator):
impl = sa.String(100)
def process_bind_param(self, value, dialect):
return ";".join(
"%s=%s" % (k, v)
for k, v in sorted(value.items(), key=lambda key: key[0])
)
def process_result_value(self, value, dialect):
return dict(elem.split("=", 1) for elem in value.split(";"))
category = Table(
"category",
metadata,
Column("id", Integer, primary_key=True),
Column("data", MyHashType()),
)
article = Table(
"article",
metadata,
Column("id", Integer, primary_key=True),
Column("data", MyHashType()),
)
class Category(fixtures.ComparableEntity):
pass
class Article(fixtures.ComparableEntity):
pass
mapper(Category, category)
mapper(
Article,
article,
properties={
"category": relationship(
Category,
primaryjoin=orm.foreign(article.c.data) == category.c.data,
load_on_pending=load_on_pending,
)
},
)
metadata.create_all()
sess = Session(autoflush=False)
data = {"im": "unhashable"}
a1 = Article(id=1, data=data)
c1 = Category(id=1, data=data)
if load_on_pending:
sess.add(c1)
else:
sess.add_all([c1, a1])
sess.flush()
if load_on_pending:
sess.add(a1)
return Category, Article, sess, a1, c1
开发者ID:vrajmohan,项目名称:sqlalchemy,代码行数:58,代码来源:test_lazy_relations.py
示例10: rel_acl
def rel_acl(self):
"""
Relation to AclEntry
:return: (object) SQLAlchemy relationship description
:since: v0.2.00
"""
return relationship(AclEntry, primaryjoin = (foreign(self.id) == remote(AclEntry.owned_id)), uselist = True)
开发者ID:dNG-git,项目名称:pas_database_ownable,代码行数:9,代码来源:ownable_mixin.py
示例11: rel_text_entry
def rel_text_entry(self):
"""
Relation to TextEntry
:return: (object) SQLAlchemy relationship description
:since: v0.2.00
"""
return relationship(TextEntry, primaryjoin = (foreign(self.id) == remote(TextEntry.id)), uselist = False)
开发者ID:dNG-git,项目名称:pas_database_text,代码行数:9,代码来源:text_mixin.py
示例12: _join_fixture_o2m_to_annotated_func
def _join_fixture_o2m_to_annotated_func(self, **kw):
return relationships.JoinCondition(
self.left,
self.right,
self.left,
self.right,
primaryjoin=self.left.c.id == foreign(func.foo(self.right.c.lid)),
**kw
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:9,代码来源:test_rel_fn.py
示例13: _join_fixture_purely_single_m2o
def _join_fixture_purely_single_m2o(self, **kw):
return relationships.JoinCondition(
self.purely_single_col,
self.purely_single_col,
self.purely_single_col,
self.purely_single_col,
support_sync=False,
primaryjoin=remote(self.purely_single_col.c.path).like(
foreign(self.purely_single_col.c.path.concat("%"))
),
)
开发者ID:BY-jk,项目名称:sqlalchemy,代码行数:11,代码来源:test_rel_fn.py
示例14: _join_fixture_remote_local_multiple_ref
def _join_fixture_remote_local_multiple_ref(self, **kw):
fn = lambda a, b: ((a == b) | (b == a))
return relationships.JoinCondition(
self.selfref, self.selfref,
self.selfref, self.selfref,
support_sync=False,
primaryjoin=fn(
# we're putting a do-nothing annotation on
# "a" so that the left/right is preserved;
# annotation vs. non seems to affect __eq__ behavior
self.selfref.c.sid._annotate({"foo": "bar"}),
foreign(remote(self.selfref.c.sid)))
)
开发者ID:CyberCollins,项目名称:sqlalchemy,代码行数:13,代码来源:test_rel_fn.py
示例15: related
def related(cls):
id_column = getattr(cls, id_label)
related_id_column = getattr(related_model, related_id_label)
filters = [foreign(id_column) == related_id_column]
if cycle_label:
cycle_column = getattr(cls, cycle_label)
if use_modulus:
cycle_column = cycle_column + cycle_column % 2
related_cycle_column = getattr(related_model, related_cycle_label)
filters.append(cycle_column == related_cycle_column)
return db.relationship(
related_model,
primaryjoin=sa.and_(*filters),
)
开发者ID:18F,项目名称:openFEC,代码行数:14,代码来源:utils.py
示例16: setup_mappers
def setup_mappers(cls):
mapper(cls.classes.Person, cls.tables.person, properties=dict(
pets=relationship(
cls.classes.Pet, primaryjoin=(
orm.foreign(cls.tables.pets.c.person_id) ==
sa.cast(
sa.type_coerce(cls.tables.person.c.id, Integer),
Integer
)
)
)
))
mapper(cls.classes.Pet, cls.tables.pets)
开发者ID:cpcloud,项目名称:sqlalchemy,代码行数:14,代码来源:test_lazy_relations.py
示例17: fetch_users_to_sync
def fetch_users_to_sync(session, required_property=None):
# type: (...) -> List[_ResultProxyType]
"""Fetch the users who should be synced
:param session: The SQLAlchemy session to use
:param str required_property: the property required to export users
:returns: An iterable of `(User, should_be_blocked)` ResultProxies
having the property `required_property' and a unix_account.
"""
if required_property:
no_unix_account_q = User.q.join(User.current_properties)\
.filter(CurrentProperty.property_name == required_property,
User.unix_account == None)
else:
no_unix_account_q = User.q.filter(User.unix_account == None)
count_exportable_but_no_account = no_unix_account_q.count()
if required_property:
logger.warning("%s users have the '%s' property but not a unix_account",
count_exportable_but_no_account, required_property)
else:
logger.warning("%s users applicable to exporting don't have a unix_account",
count_exportable_but_no_account)
# used for second join against CurrentProperty
not_blocked_property = CurrentProperty.__table__.alias('ldap_login_enabled')
return (
# Grab all users with the required property
User.q
.options(joinedload(User.unix_account))
.join(User.current_properties)
.filter(CurrentProperty.property_name == required_property,
User.unix_account != None)
# additional info:
# absence of `ldap_login_enabled` property → should_be_blocked
.add_column(not_blocked_property.c.property_name.is_(None)
.label('should_be_blocked'))
.outerjoin(
not_blocked_property,
and_(User.id == foreign(not_blocked_property.c.user_id),
~not_blocked_property.c.denied,
not_blocked_property.c.property_name == 'ldap_login_enabled')
).all()
)
开发者ID:agdsn,项目名称:pycroft,代码行数:48,代码来源:exporter.py
示例18: __declare_last__
def __declare_last__(cls):
from .invoice import Job
owner_type = PersonType.get(cls, None)
if owner_type:
cls.jobs = relation('Job',
primaryjoin=and_(
cls.guid == foreign(Job.owner_guid),
owner_type == Job.owner_type,
),
cascade='all, delete-orphan',
collection_class=CallableList,
)
@event.listens_for(cls.jobs, "append")
def add(target, value, initiator):
value.owner_type = owner_type
value.owner_guid = target.guid
value._assign_id()
开发者ID:sdementen,项目名称:piecash,代码行数:18,代码来源:person.py
示例19: relationship
Chat.request = relationship(Request)
Message.chat = relationship(Chat, backref="messages")
Message.user = relationship(User, backref="messages")
Message.chat_user = relationship(
ChatUser,
primaryjoin=and_(Message.chat_id == ChatUser.chat_id, Message.user_id == ChatUser.user_id),
foreign_keys=[Message.chat_id, Message.user_id],
)
ChatUser.chat = relationship(Chat, backref="users")
ChatUser.user = relationship(User, backref="chats")
ChatUser.export = relationship(
ChatExport,
primaryjoin=and_(
ChatUser.chat_id == foreign(ChatExport.chat_id),
ChatUser.user_id == foreign(ChatExport.user_id),
),
uselist=False,
viewonly=True,
)
ChatExport.chat = relationship(Chat)
ChatExport.user = relationship(User)
PromptReport.duplicate_of = relationship(PromptReport, backref="duplicates", remote_side=PromptReport.id)
PromptReport.reporting_user = relationship(User, backref="reports_sent", primaryjoin=PromptReport.reporting_user_id == User.id)
PromptReport.reported_user = relationship(User, backref="reports_recieved", primaryjoin=PromptReport.reported_user_id == User.id)
Request.user = relationship(User, backref="requests")
Request.tags = relationship(Tag, secondary=RequestTag.__table__, order_by=(Tag.type, Tag.name), backref="requests")
开发者ID:tehdragonfly,项目名称:cherubplay,代码行数:31,代码来源:__init__.py
示例20: mapper
if session:
session.commit()
mapper(SubNode, subnode_table,
properties=dict(_state=subnode_table.c.state))
mapper(Node, node_table,
properties=dict(
_state=node_table.c.state,
subnodes=relationship(
SubNode,
cascade='all, delete-orphan',
uselist=True,
primaryjoin=foreign(subnode_table.c.node_id) == node_table.c.id,
backref='node')))
mapper(SnapshotImage, snapshot_image_table,
properties=dict(_state=snapshot_image_table.c.state))
mapper(DibImage, dib_image_table,
properties=dict(_state=dib_image_table.c.state))
class NodeDatabase(object):
def __init__(self, dburi):
engine_kwargs = dict(echo=False, pool_recycle=3600)
if 'sqlite:' not in dburi:
engine_kwargs['max_overflow'] = -1
开发者ID:teselkin,项目名称:nodepool,代码行数:31,代码来源:nodedb.py
注:本文中的sqlalchemy.orm.foreign函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论