Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_inheritance():
"""Test type and context inheritance."""
types = {'prov:Activity', 'wfprov:ProcessRun'}
context_keys = {'prov', 'wfprov', '@version'}
@jsonld.s(type='prov:Activity', context={'prov': 'A'})
class Activity:
"""Define an activity."""
@jsonld.s(type='wfprov:ProcessRun', context={'wfprov': 'B'})
class ProcessRun(Activity):
"""Define a process execution based on an activity."""
data = jsonld.asjsonld(ProcessRun())
assert set(data['@type']) == types
assert set(data['@context'].keys()) == context_keys
types = {'prov:Activity', 'wfprov:ProcessRun', 'wfprov:WorkflowRun'}
@jsonld.s(type='wfprov:WorkflowRun')
class WorkflowRun(ProcessRun):
"""Define a workflow run."""
@jsonld.s(type='prov:Activity', context={'prov': 'A'})
class Activity:
"""Define an activity."""
@jsonld.s(type='wfprov:ProcessRun', context={'wfprov': 'B'})
class ProcessRun(Activity):
"""Define a process execution based on an activity."""
data = jsonld.asjsonld(ProcessRun())
assert set(data['@type']) == types
assert set(data['@context'].keys()) == context_keys
types = {'prov:Activity', 'wfprov:ProcessRun', 'wfprov:WorkflowRun'}
@jsonld.s(type='wfprov:WorkflowRun')
class WorkflowRun(ProcessRun):
"""Define a workflow run."""
data = jsonld.asjsonld(WorkflowRun())
assert set(data['@type']) == types
assert set(data['@context'].keys()) == context_keys
"""Represent a process."""
_activity = jsonld.ib(
context='prov:activity',
kw_only=True,
converter=weakref.ref,
type='renku.core.models.provenance.activities.Activity'
)
@property
def activity(self):
"""Return the activity object."""
return self._activity()
@jsonld.s(
type=[
'wfdesc:Workflow',
'prov:Entity',
'prov:Plan',
],
context={
'wfdesc': 'http://purl.org/wf4ever/wfdesc#',
'prov': 'http://www.w3.org/ns/prov#',
},
cmp=False,
)
class Workflow(Process):
"""Represent workflow with subprocesses."""
subprocesses = jsonld.ib(context='wfdesc:hasSubProcess', kw_only=True)
"""Convert language object."""
if isinstance(obj, dict):
language = Language.from_jsonld(obj)
return language
def _convert_keyword(keywords):
"""Convert keywords collection."""
if isinstance(keywords, list):
return keywords
if isinstance(keywords, dict):
return keywords.keys()
@jsonld.s(
type='schema:Dataset',
context={
'schema': 'http://schema.org/',
},
)
class Dataset(Entity, CreatorMixin):
"""Repesent a dataset."""
SUPPORTED_SCHEMES = ('', 'file', 'http', 'https', 'git+https', 'git+ssh')
EDITABLE_FIELDS = [
'creator', 'date_published', 'description', 'in_language', 'keywords',
'license', 'name', 'url', 'version', 'created', 'files', 'short_name'
]
_id = jsonld.ib(default=None, context='@id', kw_only=True)
)
return entity
@property
def parent(self): # pragma: no cover
"""Return the parent object."""
return self._parent() if self._parent is not None else None
@property
def entities(self):
"""Yield itself."""
yield self
@jsonld.s(
type=[
'prov:Collection',
],
context={
'prov': 'http://www.w3.org/ns/prov#',
},
cmp=False,
)
class Collection(Entity):
"""Represent a directory with files."""
members = jsonld.ib(context='prov:hadMember', kw_only=True)
@members.default
def default_members(self):
"""Generate default members as entities from current path."""
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Represent provenance agents."""
import configparser
import re
from attr.validators import instance_of
from renku.core import errors
from renku.core.models import jsonld as jsonld
from renku.version import __version__, version_url
@jsonld.s(
type=[
'prov:Person',
'schema:Person',
],
context={
'schema': 'http://schema.org/',
'prov': 'http://www.w3.org/ns/prov#',
'rdfs': 'http://www.w3.org/2000/01/rdf-schema#'
},
slots=True,
)
class Person:
"""Represent a person."""
name = jsonld.ib(
context='schema:name', kw_only=True, validator=instance_of(str)
class EntityProxyMixin:
"""Implement proxy to entity attribute."""
def __getattribute__(self, name):
"""Proxy entity attributes."""
cls = object.__getattribute__(self, '__class__')
names = {field.name for field in attr.fields(cls)}
names |= set(dir(cls))
if name in names:
return object.__getattribute__(self, name)
entity = object.__getattribute__(self, 'entity')
return getattr(entity, name)
@jsonld.s(
type='prov:Usage',
context={
'prov': 'http://www.w3.org/ns/prov#',
},
cmp=False,
)
class Usage(EntityProxyMixin):
"""Represent a dependent path."""
entity = jsonld.ib(
context='prov:entity',
kw_only=True,
type=[
'renku.core.models.entities.Entity',
'renku.core.models.entities.Collection', Dataset, DatasetFile
]
name = jsonld.ib(default=None, kw_only=True, context='schema:name')
def convert_filename_path(p):
"""Return name of the file."""
if p:
return Path(p).name
def convert_based_on(v):
"""Convert based_on to DatasetFile."""
if v:
return DatasetFile.from_jsonld(v)
@jsonld.s(
type='schema:DigitalDocument',
slots=True,
context={
'schema': 'http://schema.org/',
}
)
class DatasetFile(Entity, CreatorMixin):
"""Represent a file in a dataset."""
added = jsonld.ib(
converter=parse_date, context='schema:dateCreated', kw_only=True
)
checksum = attr.ib(default=None, kw_only=True)
filename = attr.ib(kw_only=True, converter=convert_filename_path)
@property
def creators_csv(self):
"""Comma-separated list of creators associated with dataset."""
return ','.join(creator.name for creator in self.creator)
def _extract_doi(value):
"""Return either a string or the doi part of a URL."""
value = str(value)
if is_doi(value):
return extract_doi(value)
return value
@jsonld.s(
type='schema:PublicationEvent',
context={'schema': 'http://schema.org/'},
frozen=True,
slots=True,
)
class DatasetTag(object):
"""Represents a Tag of an instance of a dataset."""
client = attr.ib(default=None, kw_only=True)
name = jsonld.ib(
default=None,
kw_only=True,
validator=instance_of(str),
context='schema:name'
)