How to use the sqlalchemy.orm.mapper function in SQLAlchemy

To help you get started, we’ve selected a few SQLAlchemy examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github sqlalchemy / sqlalchemy / test / orm / test_selectin_relations.py View on Github external
)

        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,
github sqlalchemy / sqlalchemy / test / ext / test_serializer.py View on Github external
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()
github sqlalchemy / sqlalchemy / test / orm / test_froms.py View on Github external
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",
github sqlalchemy / sqlalchemy / test / orm / inheritance / test_single.py View on Github external
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 "
github obeattie / sqlalchemy / test / orm / test_collection.py View on Github external
    @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']))
github sqlalchemy / sqlalchemy / test / orm / test_eager_relations.py View on Github external
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()
github sqlalchemy / sqlalchemy / test / ext / test_associationproxy.py View on Github external
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)
github sqlalchemy / sqlalchemy / test / orm / test_events.py View on Github external
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])
github galaxyproject / galaxy / lib / galaxy / model / mapping.py View on Github external
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)),
github pulse-project / pulse2 / services / pulse2 / database / imaging / __init__.py View on Github external
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)