Source code for invenio_sip2.records.record
# -*- coding: utf-8 -*-
#
# INVENIO-SIP2
# Copyright (C) 2020 UCLouvain
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, version 3 of the License.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""API for manipulating the client."""
from copy import deepcopy
from datetime import datetime
from uuid import uuid4
from invenio_sip2 import datastore
from ..errors import ServerAlreadyRunning
[docs]class Server(Sip2RecordMetadata):
"""class for SIP2 server."""
record_type = 'server'
@property
def number_of_clients(self):
"""Shortcut for number of clients."""
return len(self.get_clients())
@property
def is_running(self):
"""Check if server is running."""
return self.get('status') == 'running'
[docs] def delete(self):
"""Delete server and all attached clients."""
self.clear_all_clients()
super().delete()
[docs] def get_clients(self):
"""Return clients."""
filter_query = 'server:{server_id}'.format(
server_id=self.id
)
return self.search(
index_type=Client.record_type,
filter_query=filter_query
)
[docs] def down(self):
"""Set server status to `Down` and clear all clients data."""
self['status'] = 'down'
self['stopped_at'] = datetime.utcnow().isoformat()
try:
del self['process_id']
except KeyError:
pass
self.update(self)
# clear all clients
self.clear_all_clients()
[docs] def up(self):
"""Set server status to `running` and clear all clients data."""
self['status'] = 'running'
self['started_at'] = datetime.utcnow().isoformat()
self.update(self)
[docs] def clear_all_clients(self):
"""Clear all clients."""
for client in self.get_clients():
Client(client).delete()
[docs] @classmethod
def create(cls, data, id_=None, **kwargs):
"""Create record.
:param data: Dict with metadata.
:param id_: Specify a UUID to use for the new record.
"""
# check if server already exist in datastore
server = cls.find_server(**data)
if server:
# check if server running
if server.is_running:
raise ServerAlreadyRunning(
'server already running {id}'.format(id=server.id)
)
return server
return super().create(data, id_=id_)
[docs] @classmethod
def find_server(cls, **kwargs):
"""Find server depending kwargs."""
try:
del kwargs['process_id']
except KeyError:
pass
for server in datastore.all(cls.record_type):
if kwargs.items() <= server.items():
# true only if `first` is a subset of `second`
return cls(server)
[docs]class Client(Sip2RecordMetadata):
"""class for selfcheck client."""
record_type = 'client'
[docs] def get_key(self):
"""Get generated key for Client object."""
return '{record_type}:{id}_server:{server_id}'.format(
record_type=self.record_type,
id=self.id,
server_id=self.server_id
)
@property
def server_id(self):
"""Get server identifier."""
return self.get('server').get('id')
[docs] def get_server(self):
"""Get server object."""
return Server.get_record_by_id(self.server_id)
@property
def remote_app(self):
"""Shortcut for remote app."""
return self.get_server().get('remote_app')
@property
def is_authenticated(self):
"""Shortcut to check if the selfcheck client is authenticated."""
return self.get('authenticated', False)
@property
def terminal(self):
"""Shortcut to terminal."""
return self.get('terminal', self.get('ip_address'))
@property
def transaction_user_id(self):
"""Shortcut to user id."""
return self.get('transaction_user_id')
@property
def institution_id(self):
"""Shortcut to institution id."""
return self.get('institution_id')
@property
def library_name(self):
"""Shortcut to library name."""
return self.get('library_name')
[docs] def get_current_patron_session(self):
"""Shortcut to patron session."""
return self.get('patron_session', None)
[docs] def clear_patron_session(self):
"""Shortcut to library name."""
del(self['patron_session'])
@property
def last_response_message(self):
"""Shortcut to user id."""
return self.get('last_response', {})
@property
def last_request_message(self):
"""Shortcut to user id."""
return self.get('last_request', {})
@property
def get_last_sequence_number(self):
"""Shortcut to user id."""
return self.last_request_message.get('sequence_number')