Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def refresh_commit_info(self, oid, seen, lazy=True):
from allura.model.repository import CommitDoc
ci_doc = CommitDoc.m.get(_id=oid)
if ci_doc and lazy:
return False
ci = self._git.rev_parse(oid)
args = dict(
tree_id=ci.tree.hexsha,
committed=Object(
name=h.really_unicode(ci.committer.name),
email=h.really_unicode(ci.committer.email),
date=datetime.utcfromtimestamp(ci.committed_date)),
authored=Object(
name=h.really_unicode(ci.author.name),
email=h.really_unicode(ci.author.email),
date=datetime.utcfromtimestamp(ci.authored_date)),
message=h.really_unicode(ci.message or ''),
child_ids=[],
parent_ids=[p.hexsha for p in ci.parents])
if ci_doc:
ci_doc.update(**args)
ci_doc.m.save()
else:
ci_doc = CommitDoc(dict(args, _id=ci.hexsha))
try:
def get_changes(self, oid):
rev = self._revision(oid)
try:
log_entry = self._svn.log(
self._url,
revision_start=rev,
limit=1,
discover_changed_paths=True)[0]
except pysvn.ClientError:
log.info('ClientError processing %r %r, treating as empty',
oid, self._repo, exc_info=True)
log_entry = Object(date='', message='', changed_paths=[])
return [p.path for p in log_entry.changed_paths]
def create(self, doc, options, remake=True):
if remake is True or type(doc) is not self.collection:
# When querying, the ODMCursor already receives data from ming.Cursor
# which already constructed and validated the documents as collections.
# So it will leverage the remake=False option to avoid re-validating
doc = self.collection.make(doc)
mapper = self.by_collection(type(doc))
return mapper._from_doc(doc, Object(self.options, **options), validate=False)
def by_name(self):
d = Object((x.name, x) for x in self.other_ids)
d.update(
(x.name, Object(x, type='tree'))
for x in self.tree_ids)
d.update(
(x.name, Object(x, type='blob'))
for x in self.blob_ids)
return d
import logging
import pickle
from pylons import c
from ming.base import Object
from ming import schema as S
from ming.utils import LazyProperty
from allura.lib import helpers as h
log = logging.getLogger(__name__)
class ArtifactReference(Object):
"A pickle-able reference to an :class:`Artifact `"
@LazyProperty
def cls(self):
try:
return pickle.loads(str(self.artifact_type))
except:
log.exception('Error unpickling artifact reference class')
return None
@LazyProperty
def artifact(self):
if self.artifact_type is None: return None
if self.cls is None: return None
with h.push_context(self.project_id, self.mount_point):
return self.cls.query.get(_id=self.artifact_id)
def refresh_tree_info(self, tree, seen, lazy=True):
from allura.model.repository import TreeDoc
if lazy and tree.binsha in seen:
return
seen.add(tree.binsha)
doc = TreeDoc(dict(
_id=tree.hexsha,
tree_ids=[],
blob_ids=[],
other_ids=[]))
for o in tree:
if o.type == 'submodule':
continue
obj = Object(
name=h.really_unicode(o.name),
id=o.hexsha)
if o.type == 'tree':
self.refresh_tree_info(o, seen, lazy)
doc.tree_ids.append(obj)
elif o.type == 'blob':
doc.blob_ids.append(obj)
else:
obj.type = o.type
doc.other_ids.append(obj)
doc.m.save(safe=False)
return doc
def if_missing(self):
return BaseObject(
(k, v.validate(Missing))
for k,v in six.iteritems(self.fields)
if isinstance(k, six.string_types))
def clear_reason(ace):
return Object(access=ace.access, role_id=ace.role_id, permission=ace.permission)
def __init__(self, mapper, instance, options):
self.mapper = mapper
self.instance = instance
self.state = ObjectState(options, None)
self.state.document = Object()
self.state.original_document = Object()
def _safe_bson(obj, _no_warning=False):
'''Verify that the obj is safe for bsonification (in particular, no tuples or
Decimal objects
'''
if not _no_warning:
warnings.warn("_safe_bson is now deprecated. "
"If your code relies on _safe_bson for validation, "
"you should consider adding your own layer of validation", DeprecationWarning)
if isinstance(obj, list):
return [ _safe_bson(o, True) for o in obj ]
elif isinstance(obj, dict):
return Object((k, _safe_bson(v, True)) for k,v in six.iteritems(obj))
elif isinstance(obj, six.string_types + six.integer_types + (
float, datetime, NoneType,
bson.ObjectId)):
return obj
elif isinstance(obj, decimal.Decimal):
return Decimal128(obj)
else:
assert False, '%s is not safe for bsonification: %r' % (
type(obj), obj)