Source code for pyPreservica.authorityAPI
"""
pyPreservica ControlledVocabularyAPI module definition
A client library for the Preservica Repository web services Webhook API
https://us.preservica.com/api/reference-metadata/documentation.html
author: James Carr
licence: Apache License 2.0
"""
import csv
import json
from typing import List, Set
from pyPreservica.common import *
logger = logging.getLogger(__name__)
BASE_ENDPOINT = '/api/reference-metadata'
[docs]
class Table:
def __init__(self, name: str, security_tag: str, displayField: str=None, metadataConnections: list=[]):
self.reference = None
self.name = name
self.description = None
self.security_tag = security_tag
self.displayField = displayField
self.metadataConnections = metadataConnections
self.fields = None
def __str__(self):
table_str: str = f"Ref:\t\t\t{self.reference}\n" \
f"Name:\t\t\t{self.name}\n" \
f"Security Tag:\t{self.security_tag}\n"
if self.description is not None:
table_str = table_str + f"Description:\t\t\t{self.description}\n"
if self.displayField is not None:
table_str = table_str + f"Display Field:\t\t\t{self.displayField}\n"
if self.metadataConnections is not None:
table_str = table_str + f"Metadata Connections:\t\t\t{self.metadataConnections}\n"
if self.fields is not None:
table_str = table_str + f"Fields:\t\t\t{self.fields}\n"
return table_str
[docs]
class AuthorityAPI(AuthenticatedAPI):
[docs]
def delete_record(self, reference: str):
"""
Delete a record from a table by its reference
:param reference: The reference of the record to delete
:type: reference: str
"""
headers = {HEADER_TOKEN: self.token, 'accept': 'application/json;charset=UTF-8'}
response = self.session.delete(f'{self.protocol}://{self.server}{BASE_ENDPOINT}/records/{reference}',
headers=headers)
if response.status_code == requests.codes.no_content:
return None
else:
exception = HTTPException("", response.status_code, response.url, "delete_record",
response.content.decode('utf-8'))
logger.error(exception)
raise exception
[docs]
def add_records(self, table: Table, csv_file, encoding=None):
"""
Add new records to an existing table from a CSV document
:param table: The Table to add the record to
:type: table: Table
:param csv_file: The path to the CSV document
:type: csv_file: str
:param encoding: The encoding used to open the csv document
:type: encoding: str
"""
with open(csv_file, newline='', encoding=encoding) as csvfile:
reader = csv.DictReader(csvfile)
for row in reader:
if ('ID' in row) or ('id' in row):
pass
else:
row['id'] = reader.line_num
self.add_record(table, row)
[docs]
def add_record(self, table: Table, record: dict):
"""
Add a new record to an existing table
:param table: The Table to add the record to
:type: table: Table
:param record: The record as a dictionary
:type: record: dict
:return: A single record
:rtype: dict
"""
headers = {HEADER_TOKEN: self.token, 'accept': 'application/json;charset=UTF-8'}
body = {"securityDescriptor": "open", "fieldValues": []}
for key, val in record.items():
body["fieldValues"].append({"name": str(key).lower(), "value": val})
response = self.session.post(f'{self.protocol}://{self.server}{BASE_ENDPOINT}/tables/{table.reference}/records',
headers=headers, json=body)
if response.status_code == requests.codes.created:
return str(response.content.decode('utf-8'))
else:
exception = HTTPException("", response.status_code, response.url, "add_record",
response.content.decode('utf-8'))
logger.error(exception)
raise exception
[docs]
def record(self, reference: str) -> dict:
"""
Return a record by its reference
:param reference: The reference of the record
:type: reference: str
:return: A single record
:rtype: dict
"""
headers = {HEADER_TOKEN: self.token, 'accept': 'application/json;charset=UTF-8'}
response = self.session.get(f'{self.protocol}://{self.server}{BASE_ENDPOINT}/records/{reference}',
headers=headers)
if response.status_code == requests.codes.ok:
json_response = str(response.content.decode('utf-8'))
return json.loads(json_response)
else:
exception = HTTPException("", response.status_code, response.url, "record",
response.content.decode('utf-8'))
logger.error(exception)
raise exception
[docs]
def records(self, table: Table) -> List[dict]:
"""
Return all records from a table
:param table: The authority table to return the records from
:type: table: Table
:return: List of records
:rtype: list[dict]
"""
headers = {HEADER_TOKEN: self.token, 'accept': 'application/json;charset=UTF-8'}
response = self.session.get(f'{self.protocol}://{self.server}{BASE_ENDPOINT}/tables/{table.reference}/records',
headers=headers, params={"expand": "true"})
if response.status_code == requests.codes.ok:
json_response = str(response.content.decode('utf-8'))
return json.loads(json_response)['records']
else:
exception = HTTPException("", response.status_code, response.url, "records",
response.content.decode('utf-8'))
logger.error(exception)
raise exception
[docs]
def add_table(self, new_table: Table):
"""
Add a new authority table
:return: An authority table
:rtype: Table
"""
headers = {HEADER_TOKEN: self.token, 'accept': 'application/json;charset=UTF-8', 'Content-Type': 'application/json'}
table_data = {"name": new_table.name}
if new_table.description is not None:
table_data['description'] = new_table.description
if new_table.security_tag is not None:
table_data['securityDescriptor'] = new_table.security_tag
if new_table.displayField is not None:
table_data['displayField'] = new_table.displayField
if new_table.metadataConnections is not None:
table_data['metadataConnections'] = new_table.metadataConnections
if new_table.fields is not None:
table_data['fields'] = new_table.fields
response = self.session.post(f'{self.protocol}://{self.server}{BASE_ENDPOINT}/tables', data=json.dumps(table_data), headers=headers)
if response.status_code == requests.codes.created:
json_response = str(response.content.decode('utf-8'))
doc = json.loads(json_response)
return self.table(doc['ref'])
else:
exception = HTTPException("", response.status_code, response.url, "add_table",
response.content.decode('utf-8'))
logger.error(exception)
raise exception
return None
[docs]
def table(self, reference: str) -> Table:
"""
fetch an authority table by its reference
:param reference: The reference for the authority table
:type: reference: str
:return: An authority table of interest
:rtype: Table
"""
headers = {HEADER_TOKEN: self.token, 'accept': 'application/json;charset=UTF-8'}
response = self.session.get(f'{self.protocol}://{self.server}{BASE_ENDPOINT}/tables/{reference}',
headers=headers)
if response.status_code == requests.codes.ok:
json_response = str(response.content.decode('utf-8'))
doc = json.loads(json_response)
table = Table(doc['name'], doc['securityDescriptor'], doc.get('displayField', None), doc.get('metadataConnections', None))
table.reference = doc['ref']
if 'fields' in doc:
table.fields = doc['fields']
if 'description' in doc:
table.fields = doc['description']
return table
else:
exception = HTTPException("", response.status_code, response.url, "table",
response.content.decode('utf-8'))
logger.error(exception)
raise exception
[docs]
def tables(self) -> Set[Table]:
"""
List reference metadata tables
:return: Set of authority tables
:rtype: set(Table)
"""
headers = {HEADER_TOKEN: self.token, 'accept': 'application/json;charset=UTF-8'}
response = self.session.get(f'{self.protocol}://{self.server}{BASE_ENDPOINT}/tables', headers=headers)
if response.status_code == requests.codes.ok:
json_response = str(response.content.decode('utf-8'))
doc = json.loads(json_response)
results = set()
for table in doc['tables']:
t = Table(table['name'], table['securityDescriptor'], table.get('displayField', None), table.get('metadataConnections', None))
t.reference = table['ref']
if 'description' in table:
t.description = table['description']
results.add(t)
return results
else:
exception = HTTPException("", response.status_code, response.url, "tables",
response.content.decode('utf-8'))
logger.error(exception)
raise exception