Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class B(Document):
shard_b = ReferenceField(Shard)
ref = ReferenceField(A)
class C(Document):
shard_c = ReferenceField(Shard)
ref_a = ReferenceField(A)
class D(Document):
shard_d = ReferenceField(Shard)
ref_c = ReferenceField(C)
ref_a = ReferenceField(A)
class E(Document):
shard_e = ReferenceField(Shard)
refs_a = SafeReferenceListField(ReferenceField(A))
ref_b = SafeReferenceField(B)
class F(Document):
shard_f = ReferenceField(Shard)
ref_a = ReferenceField(A)
A.drop_collection()
B.drop_collection()
C.drop_collection()
D.drop_collection()
E.drop_collection()
F.drop_collection()
self.Shard = Shard
self.A = A
def test_reference_class_with_abstract_parent(self):
"""Ensure that a class with an abstract parent can be referenced.
"""
class Sibling(Document):
name = StringField()
meta = {"abstract": True}
class Sister(Sibling):
pass
class Brother(Sibling):
sibling = ReferenceField(Sibling)
Sister.drop_collection()
Brother.drop_collection()
sister = Sister(name="Alice")
sister.save()
brother = Brother(name="Bob", sibling=sister)
brother.save()
self.assertEqual(Brother.objects[0].sibling.name, sister.name)
joined = DateTimeField(db_field='j') # date joined alliance
@property
def characters(self):
return EVECharacter.objects(corporation=self)
class EVECharacter(EVEEntity):
meta = dict(
indexes = [
'owner',
],
)
alliance = ReferenceField(EVEAlliance)
corporation = ReferenceField(EVECorporation)
race = StringField(db_field='ra')
bloodline = StringField(db_field='bl')
ancestry = StringField(db_field='an')
gender = StringField(db_field='g')
security = FloatField(db_field='sec')
titles = ListField(StringField(), db_field='ti', default=list)
roles = ListField(StringField(), db_field='ro', default=list)
credentials = ListField(ReferenceField(EVECredential, reverse_delete_rule=PULL), db_field='e', default=list)
owner = ReferenceField('User', db_field='o', reverse_delete_rule=NULLIFY)
@property
def tags(self):
log = ReferenceField(Log, required=True)
status = StringField(required=True)
status_state = StringField(required=True)
name = StringField(required=True)
parent_process = StringField()
user_type = StringField()
start_type = StringField(required=True)
start_host = ReferenceField(Host, required=True)
start_user = StringField()
start_rat = ReferenceField(Rat)
start_password = StringField()
start_path = StringField()
reason = StringField()
stop_requested = StringField()
rat_iv_map = EmbeddedDocumentListField(IVOB, default=list)
known_credentials = ListField(ReferenceField(ObservedCredential), default=list)
known_devices = ListField(ReferenceField(ObservedDevice), default=list)
known_domains = ListField(ReferenceField(ObservedDomain), default=list)
known_files = ListField(ReferenceField(ObservedFile), default=list)
known_hosts = ListField(ReferenceField(ObservedHost), default=list)
known_rats = ListField(ReferenceField(ObservedRat), default=list)
known_schtasks = ListField(ReferenceField(ObservedSchtask), default=list)
known_shares = ListField(ReferenceField(ObservedShare), default=list)
known_timedeltas = ListField(ReferenceField(ObservedTimeDelta), default=list)
known_users = ListField(ReferenceField(ObservedUser), default=list)
known_persistence = ListField(ReferenceField(ObservedPersistence), default=list)
known_registry_keys = ListField(ReferenceField(ObservedRegKey), default=list)
known_services = ListField(ReferenceField(ObservedService, default=list))
known_processes = ListField(ReferenceField(ObservedProcess), default=list)
known_trashed = ListField(ReferenceField(Trashed), default=list)
known_os_versions = ListField(ReferenceField(ObservedOSVersion), default=list)
known_software = ListField(ReferenceField(ObservedSoftware), default=list)
def person(self):
"""The Person that this ban describes currently. Can be different than orig_person, in the event that the person
the event originally described was merged into another person."""
person = Person.objects(id=self.orig_person)
if person:
return person.first()
person_merge = PersonEvent.objects(target_ident=self.orig_person).first()
return person_merge.current_person
class BanHistory(EmbeddedDocument):
meta = dict(
allow_inheritance = True,
)
user = ReferenceField(User)
time = DateTimeField()
def display(self):
return None
def clean(self):
"""For some reason using default=datetime.utcnow() for time was not working as intended..."""
if not self.time:
self.time = datetime.utcnow()
def __repr__(self):
return str(type(self)) + "(" + self.display() + ")"
class CreateBanHistory(BanHistory):
def display(self):
return "Created Ban"
width_mm = FloatField(required=True)
area = IntField(required=True)
dominant_colours = ListField(StringField())
solidity = FloatField(required=True)
colour_names = ListField(StringField())
ratio = FloatField(required=True)
width = IntField(required=True)
height = IntField(required=True)
class ShredTags(EmbeddedDocument):
user = ReferenceField(User)
tags = ListField(StringField())
recognizable_chars = StringField()
angle = IntField(default=0)
pages = ListField(ReferenceField("Pages"))
# Immutable once imported from CV.
class Shred(Document):
id = StringField(max_length=200, default='', primary_key=True)
name = IntField(required=True)
features = EmbeddedDocumentField(Features)
tags = ListField(StringField())
contour = ListField(ListField(IntField()))
sheet = StringField(required=True)
piece_fname = StringField(required=True)
piece_in_context_fname = StringField(required=True)
mask_fname = StringField(required=True)
def _feature_tags(self):
cluster.
Cluster also contains user-generated tagging results (tags field).
"""
id = StringField(max_length=200, default='', primary_key=True)
users_count = IntField(default=0, db_field='usersCount')
users_skipped = ListField(ReferenceField(User), db_field='usersSkipped')
users_processed = ListField(ReferenceField(User),
db_field='usersProcessed')
batch = StringField(required=True)
tags = ListField(EmbeddedDocumentField(ShredTags))
members = ListField(EmbeddedDocumentField(ClusterMember))
parents = ListField(ReferenceField('Cluster'))
def __unicode__(self):
return self.id
@property
def features(self):
# TODO: persist features on creation.
return self.members[0].shred.features
def get_user_tags(self, user):
for shred_tags in self.tags:
# in some rare cases user reference from shred_tags has no pk field
if shred_tags.user.id == user.pk:
return shred_tags
return None
def __unicode__(self):
return self.id
def get_user_tags(self, user):
for shred_tags in self.tags:
if shred_tags.user.pk == user.pk:
return shred_tags
return None
class Tags(Document):
description = StringField(max_length=200, default='')
title = StringField(max_length=200, default='', primary_key=True)
usages = IntField(default=0)
shreds = ListField(ReferenceField(Shreds))
synonyms = ListField(StringField(max_length=200))
is_base = BooleanField(default=True)
category = StringField(max_length=200, default='')
created_by = ReferenceField(User, reverse_delete_rule=CASCADE)
created_at = DateTimeField(default=datetime.datetime.now)
def __unicode__(self):
return self.title
class Batches(Document):
name = StringField(primary_key=True, max_length=200)
created = DateTimeField(default=datetime.datetime.now)
import_took = IntField(default=0)
pages_processed = IntField(default=0)
shreds_created = IntField(default=0)
title = StringField(required=True, max_length=255)
creator = ReferenceField("User", required=True)
location = StringField()
slug = StringField(required=True, max_length=255)
start_date = DateField()
end_date = DateField()
start_time = TimeField()
end_time = TimeField()
short_description = StringField()
long_description = StringField()
short_description_markdown = StringField()
long_description_markdown = StringField()
published = BooleanField(required=True, default=False)
date_published = DateTimeField()
is_recurring = BooleanField(required=True, default=False)
parent_series = ReferenceField("EventSeries")
image = ReferenceField("Image")
facebook_url = StringField()
gcal_id = StringField()
gcal_sequence = IntField()
def get_absolute_url(self):
"""Returns the URL path that points to the client-facing version of
this event.
:returns: A URL path like ``"/events/cookies-and-code"``.
:rtype: str
"""
if self.is_recurring:
return url_for('client.recurring_event',
slug=self.slug,
index=self.index)
class Cluster(Document):
"""Cluster of one or more shreds.
Shred membership described with a ClusterMember embedded document, which
contains a reference to the shred and its relative position and angle in a
cluster.
Cluster also contains user-generated tagging results (tags field).
"""
id = StringField(max_length=200, default='', primary_key=True)
users_count = IntField(default=0, db_field='usersCount')
users_skipped = ListField(ReferenceField(User), db_field='usersSkipped')
users_processed = ListField(ReferenceField(User),
db_field='usersProcessed')
batch = StringField(required=True)
tags = ListField(EmbeddedDocumentField(ShredTags))
members = ListField(EmbeddedDocumentField(ClusterMember))
parents = ListField(ReferenceField('Cluster'))
def __unicode__(self):
return self.id
@property
def features(self):
# TODO: persist features on creation.
return self.members[0].shred.features