Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
run_list=self.run_list)
def update(self, d):
if 'name' in d:
self.name = d['name']
self.chef_environment = d['chef_environment']
self.normal=dumps(d['normal'])
self.default=dumps(d['default'])
self.override=dumps(d['override'])
self.automatic=dumps(d['automatic'])
self.run_list = d['run_list']
orm_session.mapper(Node, node, properties=dict(
account_id=ForeignIdProperty('Account'),
account=RelationProperty('Account')))
project=p_shortname,
project_id=p_id,
app=None,
artifact=parts[0])
else:
return None
# Mapper definitions
mapper(ArtifactReference, ArtifactReferenceDoc, main_orm_session)
mapper(Shortlink, ShortlinkDoc, main_orm_session, properties=dict(
ref_id=ForeignIdProperty(ArtifactReference),
project_id=ForeignIdProperty('Project'),
app_config_id=ForeignIdProperty('AppConfig'),
project=RelationProperty('Project'),
app_config=RelationProperty('AppConfig'),
ref=RelationProperty(ArtifactReference)))
type_s = 'Ticket'
_id = FieldProperty(schema.ObjectId)
created_date = FieldProperty(datetime, if_missing=datetime.utcnow)
ticket_num = FieldProperty(int, required=True, allow_none=False)
summary = FieldProperty(str)
description = FieldProperty(str, if_missing='')
description_cache = FieldProperty(MarkdownCache)
reported_by_id = AlluraUserProperty(if_missing=lambda: c.user._id)
assigned_to_id = AlluraUserProperty(if_missing=None)
milestone = FieldProperty(str, if_missing='')
status = FieldProperty(str, if_missing='')
custom_fields = FieldProperty({str: None})
reported_by = RelationProperty(User, via='reported_by_id')
def link_text(self):
text = super(Ticket, self).link_text()
if self.is_closed:
return jinja2.Markup('<s>') + text + jinja2.Markup('</s>')
return text
@property
def activity_name(self):
return 'ticket #%s' % self.ticket_num
@property
def activity_extras(self):
d = ActivityObject.activity_extras.fget(self)
d.update(summary=self.summary)
return d
class Membership(MappedClass):
class __mongometa__:
session = main_orm_session
name='organization_membership'
_id=FieldProperty(S.ObjectId)
status=FieldProperty(S.OneOf('active', 'closed', 'invitation', 'request'))
role=FieldProperty(str)
organization_id=ForeignIdProperty('Organization')
member_id=ForeignIdProperty('User')
startdate = FieldProperty(S.DateTime, if_missing=None)
closeddate = FieldProperty(S.DateTime, if_missing=None)
organization = RelationProperty('Organization')
member = RelationProperty('User')
@classmethod
def insert(cls, role, status, organization_id, member_id):
m = cls.query.find(dict(organization_id=organization_id, member_id=member_id))
for el in m:
if el.status!='closed':
return None
try:
m = cls(
organization_id=organization_id,
member_id=member_id,
role=role,
startdate=None,
status=status)
session(m).flush(m)
except pymongo.errors.DuplicateKeyError:
class ForumPostHistory(M.PostHistory):
class __mongometa__:
name='post_history'
artifact_id = ForeignIdProperty('ForumPost')
class ForumPost(M.Post):
class __mongometa__:
name='forum_post'
history_class = ForumPostHistory
type_s = 'Post'
discussion_id = ForeignIdProperty(Forum)
thread_id = ForeignIdProperty(ForumThread)
discussion = RelationProperty(Forum)
thread = RelationProperty(ForumThread)
@classmethod
def attachment_class(cls):
return ForumAttachment
@property
def email_address(self):
return self.discussion.email_address
def primary(self):
return self
def promote(self):
'''Make the post its own thread head'''
thd = self.thread_class()(
def shorthand_id(self):
return self.short
class AwardGrant(Artifact):
"An :class:`Award ` can be bestowed upon a project by a neighborhood"
class __mongometa__:
session = main_orm_session
name = 'grant'
indexes = ['short']
type_s = 'Generic Award Grant'
_id = FieldProperty(S.ObjectId)
award_id = ForeignIdProperty(Award, if_missing=None)
award = RelationProperty(Award, via='award_id')
granted_by_neighborhood_id = ForeignIdProperty(
'Neighborhood', if_missing=None)
granted_by_neighborhood = RelationProperty(
'Neighborhood', via='granted_by_neighborhood_id')
granted_to_project_id = ForeignIdProperty('Project', if_missing=None)
granted_to_project = RelationProperty(
'Project', via='granted_to_project_id')
award_url = FieldProperty(str, if_missing='')
comment = FieldProperty(str, if_missing='')
timestamp = FieldProperty(datetime, if_missing=datetime.utcnow)
def index(self):
result = Artifact.index(self)
result.update(
_id_s=self._id,
short_s=self.short,
class ProjectInvolvement(MappedClass):
class __mongometa__:
session = main_orm_session
name='project_involvement'
_id=FieldProperty(S.ObjectId)
status=FieldProperty(S.OneOf('active', 'closed', 'invitation', 'request'))
collaborationtype=FieldProperty(S.OneOf('cooperation', 'participation'))
organization_id=ForeignIdProperty('Organization')
project_id=ForeignIdProperty('Project')
startdate = FieldProperty(S.DateTime, if_missing=None)
closeddate = FieldProperty(S.DateTime, if_missing=None)
organization = RelationProperty('Organization')
project = RelationProperty('Project')
@classmethod
def insert(cls, status, collaborationtype, organization_id, project_id):
p = cls.query.find(dict(
organization_id=organization_id,
project_id=project_id))
for el in p:
if p.status != 'closed':
return None
try:
m = cls(organization_id=organization_id, project_id=project_id, status=status, collaborationtype=collaborationtype)
session(m).flush(m)
except pymongo.errors.DuplicateKeyError:
session(m).expunge(m)
m = cls.query.get(organization_id=organization_id, project_id=project_id)
return m
_id=self._id, account_id=self.account_id, needs_upload=False) as ofp:
while True:
chunk = ifp.read(self.CHUNKSIZE)
if chunk: ofp.write(chunk)
else: break
if ofp.md5 != self.md5:
import pdb; pdb.set_trace()
chef_file.m.fs.delete(self._id)
raise Invalid('Invalid checksum', None, None)
orm_session.mapper(Sandbox, sandbox, properties=dict(
account_id=ForeignIdProperty('Account'),
account=RelationProperty('Account')))
orm_session.mapper(ChefFile, chef_file,properties=dict(
account_id=ForeignIdProperty('Account'),
account=RelationProperty('Account')))
summary=FieldProperty(str, if_missing='')
description=FieldProperty(str, if_missing='')
homepage_title=FieldProperty(str, if_missing='')
external_homepage=FieldProperty(str, if_missing='')
support_page=FieldProperty(str, if_missing='')
support_page_url=FieldProperty(str, if_missing='')
removal=FieldProperty(str, if_missing='')
moved_to_url=FieldProperty(str, if_missing='')
removal_changed_date = FieldProperty(datetime, if_missing=datetime.utcnow)
export_controlled=FieldProperty(bool, if_missing=False)
database=FieldProperty(S.Deprecated)
database_uri=FieldProperty(str)
is_root=FieldProperty(bool)
acl = FieldProperty(ACL(permissions=_perms_init))
neighborhood_invitations=FieldProperty([S.ObjectId])
neighborhood = RelationProperty(Neighborhood)
app_configs = RelationProperty('AppConfig')
category_id = FieldProperty(S.ObjectId, if_missing=None)
deleted = FieldProperty(bool, if_missing=False)
labels = FieldProperty([str])
last_updated = FieldProperty(datetime, if_missing=None)
tool_data = FieldProperty({str:{str:None}}) # entry point: prefs dict
ordinal = FieldProperty(int, if_missing=0)
database_configured = FieldProperty(bool, if_missing=True)
_extra_tool_status = FieldProperty([str])
trove_root_database=FieldProperty([S.ObjectId])
trove_developmentstatus=FieldProperty([S.ObjectId])
trove_audience=FieldProperty([S.ObjectId])
trove_license=FieldProperty([S.ObjectId])
trove_os=FieldProperty([S.ObjectId])
trove_language=FieldProperty([S.ObjectId])
trove_topic=FieldProperty([S.ObjectId])
fullname=FieldProperty(str)
organization_type=FieldProperty(S.OneOf(
'For-profit business',
'Foundation or other non-profit organization',
'Research and/or education institution'))
description=FieldProperty(str)
headquarters=FieldProperty(str)
dimension=FieldProperty(
S.OneOf('Small', 'Medium', 'Large', 'Unknown'),
if_missing = 'Unknown')
website=FieldProperty(str)
workfields=FieldProperty([S.ObjectId])
created=FieldProperty(S.DateTime, if_missing=datetime.utcnow())
memberships=RelationProperty('Membership')
project_involvements=RelationProperty('ProjectInvolvement')
def url(self):
return ('/o/' + self.shortname.replace('_', '-') + '/').encode('ascii','ignore')
def project(self):
return M.Project.query.get(
shortname='o/'+self.shortname.replace('_', '-'))
@classmethod
def register(cls, shortname, fullname, orgtype, user):
o=cls.query.get(shortname=shortname)
if o is not None: return None
try:
o = cls(
shortname=shortname,
fullname=fullname,