Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
),
)
}
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)
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))
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):
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"),
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))
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")
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')
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")
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