How to use the sqlalchemy.orm.relationship 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_cascade.py View on Github external
),
                    )
                }

            else:
                addresses_rel = {
                    "addresses": relationship(
                        Address,
                        cascade=o2m_cascade and "save-update" or "",
                        cascade_backrefs=o2m_cascade_backrefs,
                    )
                }
            user_rel = {}
        elif m2o:
            user_rel = {
                "user": relationship(
                    User,
                    cascade=m2o_cascade and "save-update" or "",
                    cascade_backrefs=m2o_cascade_backrefs,
                )
            }
            addresses_rel = {}
        else:
            addresses_rel = {}
            user_rel = {}

        mapper(User, users, properties=addresses_rel)
        mapper(Address, addresses, properties=user_rel)
github arskom / spyne / spyne / util / sqlalchemy.py View on Github external
child_left_col,
            child_right_col,
        )
        _gen_index_info(child_t, child_right_col, child_right_col_name, child_cust)

    # generate temporary class for association proxy
    cls_name = ''.join(x.capitalize() or '_' for x in
                                                child_table_name.split('_'))
                                                # generates camelcase class name.

    def _i(self, *args):
        setattr(self, child_right_col_name, args[0])

    cls_ = type("_" + cls_name, (object,), {'__init__': _i})
    own_mapper(cls_)(cls_, child_t)
    props["_" + k] = relationship(cls_)

    # generate association proxy
    setattr(cls, k, association_proxy("_" + k, child_right_col_name))
github c0rvax / project-black / black / db / models / project.py View on Github external
class ProjectDatabase(Base):
    __tablename__ = 'projects'

    project_uuid = Column(Integer, primary_key=True, autoincrement=True)
    project_name = Column(String)
    comment = Column(String, default='')

    # Marks whether any user has locked ips (hosts), disallowing adding new ones 
    ips_locked = Column(Boolean, default=False)
    hosts_locked = Column(Boolean, default=False)

    date_added = Column(DateTime, default=datetime.datetime.utcnow)

    ips_relationship = relationship('IPDatabase', cascade="all, delete-orphan")
    hosts_relationship = relationship('HostDatabase', cascade="all, delete-orphan")
    tasks_relationship = relationship('TaskDatabase', cascade="all, delete-orphan")
    scans_relationship = relationship('ScanDatabase', cascade="all, delete-orphan")

    session_spawner = Sessions()

    def dict(self):
        return {
            "project_uuid": self.project_uuid,
            "project_name": self.project_name,
            "comment": self.comment,
            "ips_locked": self.ips_locked,
            "hosts_locked": self.hosts_locked,
            "date_added": str(self.date_added)
        }

    @classmethod
    async def create(cls, project_name):
github qlands / FormShare / formshare / models / formshare.py View on Github external
project_id = Column(
        ForeignKey("project.project_id", ondelete="CASCADE"),
        primary_key=True,
        nullable=False,
        index=True,
    )
    access_type = Column(
        INTEGER
    )  # 1=Owner,2=Admin,3=Editor,4=Member. Note: 5=Public access (Set internally)
    access_date = Column(DateTime)
    project_active = Column(INTEGER, server_default=text("'1'"))
    project_accepted = Column(INTEGER, server_default=text("'1'"))
    project_accepted_date = Column(DateTime)

    project = relationship("Project")
    user = relationship("User")


class Collingroup(Base):
    __tablename__ = "collingroup"
    __table_args__ = (
        ForeignKeyConstraint(
            ["enum_project", "coll_id"],
            ["collaborator.project_id", "collaborator.coll_id"],
            ondelete="CASCADE",
        ),
        ForeignKeyConstraint(
            ["project_id", "group_id"],
            ["collgroup.project_id", "collgroup.group_id"],
            ondelete="CASCADE",
        ),
        Index("fk_enumingroup_enumerator1_idx", "enum_project", "coll_id"),
github assembl / assembl / assembl / models / idea.py View on Github external
doc="are messages in this idea also part of the parent idea?")

    message_view_override = Column(
        String(100), doc="Use a non-standard view for this idea")

    creation_date = Column(
        DateTime, nullable=False, default=datetime.utcnow)

    discussion_id = Column(Integer, ForeignKey(
        'discussion.id',
        ondelete='CASCADE',
        onupdate='CASCADE'),
        nullable=False,
        index=True)

    discussion = relationship(
        Discussion,
        backref=backref(
            'ideas', order_by=creation_date,
            primaryjoin="and_(Idea.discussion_id==Discussion.id, "
                        "Idea.tombstone_date == None)")
    )

    discussion_ts = relationship(
        Discussion,
        backref=backref(
            'ideas_ts', order_by=creation_date,
            cascade="all, delete-orphan")
    )

    # widget_id = deferred(Column(Integer, ForeignKey('widget.id')))
    # widget = relationship("Widget", backref=backref('ideas', order_by=creation_date))
github abilian / abilian-sbe / abilian / sbe / apps / communities / models.py View on Github external
logger = logging.getLogger(__name__)

MEMBER = Role("member", label=_l("role_member"), assignable=False)
VALID_ROLES = frozenset([READER, WRITER, MANAGER, MEMBER])


class Membership(db.Model):
    """Represents the membership of someone in a community."""

    __tablename__ = "community_membership"

    id = Column(Integer, primary_key=True)

    user_id = Column(ForeignKey("user.id"), index=True, nullable=False)
    user = relationship(
        User, lazy="joined", backref=backref("communautes_membership", lazy="select")
    )

    community_id = Column(ForeignKey("community.id"), index=True, nullable=False)
    community = relationship("Community", lazy="joined")

    role = Column(RoleType())  # should be either 'member' or 'manager'

    __table_args__ = (UniqueConstraint("user_id", "community_id"),)

    def __repr__(self):
        return "".format(
            repr(self.user), repr(self.community), str(self.role)
        ).encode("utf-8")
github mozilla / build-relengapi / relengapi / blueprints / tooltool / tables.py View on Github external
class PendingUpload(db.declarative_base(DB_DECLARATIVE_BASE)):

    """Files for which upload URLs have been generated, but which haven't yet
    been uploaded.  This table is used to poll for completed uploads, and to
    prevent trusting files for which there is an outstanding signed upload URL."""

    __tablename__ = 'releng_tooltool_pending_upload'

    file_id = sa.Column(
        sa.Integer, sa.ForeignKey('releng_tooltool_files.id'),
        nullable=False, primary_key=True)
    expires = sa.Column(db.UTCDateTime, index=True, nullable=False)
    region = sa.Column(
        sa.Enum(*allowed_regions), nullable=False)

    file = sa.orm.relationship('File', backref='pending_uploads')
github GOVCERT-LU / ce1sus / ce1sus / brokers / event / eventclasses.py View on Github external
description = Column('description', String)
  first_seen = Column('first_seen', DateTime)
  last_seen = Column('last_seen', DateTime)
  modified = Column('modified', DateTime)
  tlp_level_id = Column('tlp_level_id', Integer)
  published = Column('published', Integer)
  status_id = Column('status_id', Integer)
  risk_id = Column('risk_id', Integer)
  analysis_status_id = Column('analysis_status_id', Integer)
  comments = relationship("Comment")
  maingroups = relationship(Group,
                            secondary='Groups_has_Events',
                            backref="events")
  subgroups = relationship(SubGroup,
                           secondary='SubGroups_has_Events')
  objects = relationship('Object',
                         primaryjoin='and_(Event.identifier==Object.event_id, Object.parent_object_id==None)')
  created = Column('created',
                   DateTime(timezone=True))
  modified = Column('modified',
                    DateTime(timezone=True))
  # creators and modifiers will be groups
  creator_id = Column('creator_id',
                      Integer,
                      ForeignKey('Users.user_id'))
  creator = relationship(User,
                         primaryjoin="Event.creator_id==User.identifier")
  modifier_id = Column('modifier_id',
                       Integer,
                       ForeignKey('Users.user_id'))
  modifier = relationship(User,
                          primaryjoin="Event.modifier_id==User.identifier")
github EIDA / mediatorws / eidangservices / stationlite / engine / orm.py View on Github external
index=True)
    description = Column(Unicode(LENGTH_DESCRIPTION))
    longitude = Column(Float, nullable=False, index=True)
    latitude = Column(Float, nullable=False, index=True)

    station = relationship('Station', back_populates='station_epochs')


class Routing(EpochMixin, LastSeenMixin, ORMBase):

    channel_epoch_ref = Column(Integer, ForeignKey('channelepoch.id'),
                               index=True)
    endpoint_ref = Column(Integer, ForeignKey('endpoint.id'),
                          index=True)

    channel_epoch = relationship('ChannelEpoch',
                                 back_populates='endpoints')
    endpoint = relationship('Endpoint',
                            back_populates='channel_epochs')

    def __repr__(self):
        return ('' %
                (self.endpoint.url, self.starttime, self.endtime))


class Endpoint(ORMBase):

    service_ref = Column(Integer, ForeignKey('service.id'),
                         index=True)
    url = Column(String(LENGTH_URL), nullable=False)

    # many to many ChannelEpoch<->Endpoint