Source code for asclepias_broker.metadata.models

# -*- coding: utf-8 -*-
#
# Copyright (C) 2018 CERN.
#
# Asclepias Broker is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.
"""Metadata database models."""

from copy import deepcopy

import jsonschema
from invenio_db import db
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import backref
from sqlalchemy.orm.attributes import flag_modified
from sqlalchemy_utils.models import Timestamp
from sqlalchemy_utils.types import JSONType, UUIDType

from ..graph.models import Group, GroupRelationship
from ..jsonschemas import SCHOLIX_SCHEMA

COMMON_SCHEMA_DEFINITIONS = SCHOLIX_SCHEMA['definitions']
OBJECT_TYPE_SCHEMA = COMMON_SCHEMA_DEFINITIONS['ObjectType']
OVERRIDABLE_KEYS = {'Type', 'Title', 'Creator', 'PublicationDate', 'Publisher'}


[docs]class GroupMetadata(db.Model, Timestamp): """Metadata for a group.""" __tablename__ = 'groupmetadata' # TODO: assert group.type == GroupType.Identity group_id = db.Column( UUIDType, db.ForeignKey(Group.id, onupdate='CASCADE', ondelete='CASCADE'), primary_key=True) group = db.relationship( Group, backref=backref('data', uselist=False), single_parent=True, ) json = db.Column( db.JSON() .with_variant(postgresql.JSONB(none_as_null=True), 'postgresql') .with_variant(JSONType(), 'sqlite'), default=dict, ) # Identifier metadata SCHEMA = { '$schema': 'http://json-schema.org/draft-06/schema#', 'definitions': COMMON_SCHEMA_DEFINITIONS, 'additionalProperties': False, 'properties': { k: v for k, v in OBJECT_TYPE_SCHEMA['properties'].items() if k in OVERRIDABLE_KEYS }, }
[docs] def update(self, payload: dict, validate: bool = True): """Update the metadata of a group.""" new_json = deepcopy(self.json or {}) for key in OVERRIDABLE_KEYS: if payload.get(key): if key == 'Type': type_val = (payload['Type'] or {}).get('Name', 'unknown') if type_val == 'unknown': continue new_json[key] = payload[key] # Set "Type" to "unknown" if not provided if not new_json.get('Type', {}).get('Name'): new_json['Type'] = {'Name': 'unknown'} if validate: jsonschema.validate(new_json, self.SCHEMA) self.json = new_json flag_modified(self, 'json') return self
[docs]class GroupRelationshipMetadata(db.Model, Timestamp): """Metadata for a group relationship.""" __tablename__ = 'grouprelationshipmetadata' # TODO: assert group_relationship.type == GroupType.Identity group_relationship_id = db.Column( UUIDType, db.ForeignKey( GroupRelationship.id, onupdate='CASCADE', ondelete='CASCADE'), primary_key=True ) group_relationship = db.relationship( GroupRelationship, backref=backref('data', uselist=False), single_parent=True, ) json = db.Column( db.JSON() .with_variant(postgresql.JSONB(none_as_null=True), 'postgresql') .with_variant(JSONType(), 'sqlite'), default=list, ) # Relationship metadata SCHEMA = { '$schema': 'http://json-schema.org/draft-06/schema#', 'definitions': COMMON_SCHEMA_DEFINITIONS, 'type': 'array', 'items': { 'type': 'object', 'additionalProperties': False, 'properties': { 'LinkPublicationDate': {'$ref': '#/definitions/DateType'}, 'LinkProvider': { 'type': 'array', 'items': {'$ref': '#/definitions/PersonOrOrgType'} }, 'LicenseURL': {'type': 'string'}, }, 'required': ['LinkPublicationDate', 'LinkProvider'], } }
[docs] def update(self, payload: dict, validate: bool = True, multi: bool = False): """Updates the metadata of a group relationship.""" new_json = deepcopy(self.json or []) if multi: new_json.extend(payload) else: new_json.append(payload) if validate: jsonschema.validate(new_json, self.SCHEMA) self.json = new_json flag_modified(self, 'json') return self