Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def dump_project(project, dirname):
if not os.path.exists(dirname):
os.mkdir(dirname)
with open(os.path.join(dirname, 'project.bson'), 'w') as fp:
_write_bson(fp, state(project).document)
c.project = project
app_config_ids = [
ac._id for ac in M.AppConfig.query.find(dict(project_id=c.project._id)) ]
visited_collections = {}
for m in Mapper.all_mappers():
cls = m.mapped_class
cname = cls.__module__ + '.' + cls.__name__
mgr = m.collection.m
sess = m.session
if sess is None:
log.info('Skipping %s which has no session', cls)
continue
dbname = sess.impl.db.name
fqname = cname + '/' + dbname
if fqname in visited_collections:
log.info('Skipping %s (already dumped collection %s in %s)',
def create_thumbnail(self, attachment, att_cls):
if attachment.is_image():
base.log.info("Processing image attachment '%s'",
attachment.filename)
doc = state(attachment).document.deinstrumented_clone()
del doc['_id']
del doc['file_id']
doc['type'] = 'thumbnail'
count = att_cls.query.find(doc).count()
if count == 1:
base.log.info(
"Thumbnail already exists for '%s' - skipping", attachment.filename)
return
elif count > 1:
base.log.warning(
"There are %d thumbnails for '%s' - consider clearing them with --force", count, attachment.filename)
return
image = PIL.Image.open(attachment.rfile())
del doc['content_type']
del doc['filename']
def set_tool_data(self, tool, **kw):
d = self.tool_data.setdefault(tool, {})
d.update(kw)
state(self).soil()
artifact_class='%s.%s' % (
self.__class__.__module__,
self.__class__.__name__),
author=dict(
id=c.user._id,
username=c.user.username,
display_name=c.user.get_pref('display_name'),
logged_ip=ip_address),
data=state(self).clone())
while True:
self.version += 1
data['version'] = self.version
data['timestamp'] = datetime.utcnow()
ss = self.__mongometa__.history_class(**data)
try:
session(ss).insert_now(ss, state(ss))
except pymongo.errors.DuplicateKeyError:
log.warning('Trying to create duplicate version %s of %s',
self.version, self.__class__)
session(ss).expunge(ss)
continue
else:
break
log.debug('Snapshot version %s of %s',
self.version, self.__class__)
if update_stats:
if self.version > 1:
g.statsUpdater.modifiedArtifact(
self.type_s, self.mod_date, self.project, c.user)
else:
g.statsUpdater.newArtifact(
self.type_s, self.mod_date, self.project, c.user)
def set_tool_data(self, tool, **kw):
d = self.tool_data.setdefault(tool, {})
d.update(kw)
state(self).soil()
def make_slug(self):
base = BlogPost.make_base_slug(self.title, self.timestamp)
self.slug = base
while True:
try:
session(self).insert_now(self, state(self))
return self.slug
except DuplicateKeyError:
self.slug = base + '-%.3d' % randint(0, 999)
def down(self):
if self.session.db.name == 'allura':
theme = self.ormsession.find(M.Theme, {'name':'forge_default'}).first()
if not theme: return
theme.color1='#104a75'
theme.color2='#aed0ea'
theme.color3='#EDF3FB'
theme.color4='#D7E8F5'
theme.color5='#000'
theme.color6='#000'
self.ormsession.update_now(theme, state(theme))
self.ormsession.flush()
def set_tool_data(self, tool, **kw):
d = self.tool_data.setdefault(tool, {})
d.update(kw)
state(self).soil()