Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
)
mapper(Address, addresses)
mapper(
Order,
orders,
properties={
"items": relationship(
Item,
secondary=order_items,
lazy="selectin",
order_by=items.c.id,
)
},
)
mapper(Item, items)
open_mapper = aliased(Order, orders)
closed_mapper = aliased(Order, orders)
mapper(
User,
users,
properties=dict(
addresses=relationship(
Address, lazy="selectin", order_by=addresses.c.id
),
open_orders=relationship(
open_mapper,
primaryjoin=sa.and_(
open_mapper.isopen == 1,
users.c.id == open_mapper.user_id,
def setup_mappers(cls):
global Session
Session = scoped_session(sessionmaker())
mapper(
User,
users,
properties={
"addresses": relationship(
Address, backref="user", order_by=addresses.c.id
)
},
)
mapper(Address, addresses)
configure_mappers()
def test_join_relname_from_selected_from(self):
User, Address = self.classes.User, self.classes.Address
users, addresses = self.tables.users, self.tables.addresses
mapper(
User,
users,
properties={
"addresses": relationship(
mapper(Address, addresses), backref="user"
)
},
)
sess = create_session()
self.assert_compile(
sess.query(User).select_from(Address).join("user"),
"SELECT users.id AS users_id, users.name AS users_name "
"FROM addresses JOIN users ON users.id = addresses.user_id",
def test_of_type_aliased_fromjoinpoint(self):
Company, Employee, Engineer = (
self.classes.Company,
self.classes.Employee,
self.classes.Engineer,
)
companies, employees = self.tables.companies, self.tables.employees
mapper(
Company, companies, properties={"employee": relationship(Employee)}
)
mapper(Employee, employees, polymorphic_on=employees.c.type)
mapper(Engineer, inherits=Employee, polymorphic_identity="engineer")
sess = create_session()
self.assert_compile(
sess.query(Company).outerjoin(
Company.employee.of_type(Engineer),
aliased=True,
from_joinpoint=True,
),
"SELECT companies.company_id AS companies_company_id, "
"companies.name AS companies_name FROM companies "
@testing.resolve_artifact_names
def _test_scalar_mapped(self, collection_class):
mapper(Child, children)
mapper(Parent, parents, properties={
'children': relation(Child, collection_class=collection_class,
cascade="all, delete-orphan")})
p = Parent()
p.children['foo'] = Child('foo', 'value')
p.children['bar'] = Child('bar', 'value')
session = create_session()
session.add(p)
session.flush()
pid = p.id
session.expunge_all()
p = session.query(Parent).get(pid)
eq_(set(p.children.keys()), set(['foo', 'bar']))
def test_options(self):
nodes = self.tables.nodes
class Node(fixtures.ComparableEntity):
def append(self, node):
self.children.append(node)
mapper(Node, nodes, properties={
'children':relationship(Node, lazy='select', order_by=nodes.c.id)
}, order_by=nodes.c.id)
sess = create_session()
n1 = Node(data='n1')
n1.append(Node(data='n11'))
n1.append(Node(data='n12'))
n1.append(Node(data='n13'))
n1.children[1].append(Node(data='n121'))
n1.children[1].append(Node(data='n122'))
n1.children[1].append(Node(data='n123'))
sess.add(n1)
sess.flush()
sess.expunge_all()
def go():
d = sess.query(Node).filter_by(data='n1').\
options(joinedload('children.children')).first()
def test_lazy_scalar(self):
Parent, Child = self.Parent, self.Child
mapper(
Parent,
self.table,
properties={
"_children": relationship(Child, lazy="select", uselist=False)
},
)
p = Parent("p")
p.children = "value"
p = self.roundtrip(p)
self.assert_("_children" not in p.__dict__)
self.assert_(p._children is not None)
users, User = (self.tables.users, self.classes.User)
class SubUser(User):
pass
class SubUser2(User):
pass
canary = []
def evt(x):
canary.append(x)
event.listen(User, "load", evt, propagate=True, raw=True)
m = mapper(SubUser, users)
m2 = mapper(SubUser2, users)
m.class_manager.dispatch.load(5)
eq_(canary, [5])
m2.class_manager.dispatch.load(5)
eq_(canary, [5, 5])
user=relation(model.User)
))
mapper(model.GalaxySession, model.GalaxySession.table, properties=dict(
histories=relation(model.GalaxySessionToHistoryAssociation),
current_history=relation(model.History),
# user=relation( model.User.mapper ) ) )
user=relation(model.User)
))
mapper(model.GalaxySessionToHistoryAssociation, model.GalaxySessionToHistoryAssociation.table, properties=dict(
galaxy_session=relation(model.GalaxySession),
history=relation(model.History)
))
mapper(model.Workflow, model.Workflow.table, properties=dict(
steps=relation(model.WorkflowStep,
backref='workflow',
primaryjoin=((model.Workflow.table.c.id == model.WorkflowStep.table.c.workflow_id)),
order_by=asc(model.WorkflowStep.table.c.order_index),
cascade="all, delete-orphan",
lazy=False),
step_count=column_property(
select([func.count(model.WorkflowStep.table.c.id)]).where(model.Workflow.table.c.id == model.WorkflowStep.table.c.workflow_id),
deferred=True
)
))
mapper(model.WorkflowStep, model.WorkflowStep.table, properties=dict(
subworkflow=relation(model.Workflow,
primaryjoin=((model.Workflow.table.c.id == model.WorkflowStep.table.c.subworkflow_id)),
cascade="all,delete-orphan")
})
mapper(ComputerPartition, self.computer_partition)
mapper(Entity, self.entity)
mapper(Image, self.image)
mapper(ImageInMenu, self.image_in_menu)
mapper(ImageOnImagingServer, self.image_on_imaging_server)
mapper(ImagingServer, self.imaging_server)
mapper(Internationalization, self.internationalization)
mapper(Language, self.language)
mapper(ImagingLog, self.imaging_log)
mapper(ImagingLogState, self.imaging_log_state)
mapper(ImagingLogLevel, self.imaging_log_level)
mapper(MasteredOn, self.mastered_on)
mapper(Menu, self.menu) #, properties = { 'default_item':relation(MenuItem), 'default_item_WOL':relation(MenuItem) } )
mapper(MenuItem, self.menu_item) #, properties = { 'menu' : relation(Menu) })
mapper(Partition, self.partition)
mapper(PostInstallScript, self.post_install_script)
mapper(PostInstallScriptInImage, self.post_install_script_in_image)
mapper(PostInstallScriptOnImagingServer, self.post_install_script_on_imaging_server)
mapper(Protocol, self.protocol)
mapper(SynchroState, self.synchro_state)
mapper(Target, self.target, properties = {
'disks' : relation(ComputerDisk,
cascade="all,delete-orphan")
})
mapper(TargetType, self.target_type)
mapper(User, self.user)