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'




class Table:
    def __init__(self, name: str, security_tag: str, displayField: str=None, metadataConnections: list=[]):
        """
        Initialise an authority Table with its name, security tag, and optional display settings.

        :param name: Human-readable name for the authority table.
        :type name: str
        :param security_tag: Security descriptor (access control tag) applied to the table
            (e.g. ``"open"``).
        :type security_tag: str
        :param displayField: The field name used as the human-readable label when records from
            this table are displayed in the UI.  ``None`` means no display field is set.
        :type displayField: str or None
        :param metadataConnections: List of metadata connection definitions that link this table
            to metadata schemas in Preservica.
        :type 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):
        """
        Return a human-readable string representation of this Table.

        :returns: Multi-line string containing the table reference, name, security tag, and
            optionally description, display field, metadata connections, and fields.
        :rtype: str
        """
        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





class AuthorityAPI(AuthenticatedAPI):



    def delete_record(self, reference: str):
        """
        Delete a record from a table by its reference.

        :param reference: The unique reference of the record to delete.
        :type reference: str
        :returns: None
        :rtype: None
        :raises HTTPException: If the Preservica API returns an unexpected HTTP error status.
        """
        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




    def add_records(self, table: Table, csv_file, encoding=None):
        """
        Add new records to an existing authority table from a CSV document.

        Each row in the CSV becomes one record.  If the CSV does not already contain an
        ``id`` or ``ID`` column, the reader's line number is used as the record ID.

        :param table: The authority table to which records will be added.
        :type table: Table
        :param csv_file: Path to the CSV file whose rows will be imported as records.
        :type csv_file: str
        :param encoding: Character encoding used to open the CSV file (e.g. ``"utf-8"``).
            Defaults to the platform default when ``None``.
        :type encoding: str or None
        :returns: None
        :rtype: None
        :raises HTTPException: If any individual ``add_record`` call returns an unexpected
            HTTP error status.
        """
        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)




    def add_record(self, table: Table, record: dict):
        """
        Add a single new record to an existing authority table.

        Dictionary keys become field names (lowercased) and dictionary values become the
        corresponding field values in the new record.

        :param table: The authority table to which the record will be added.
        :type table: Table
        :param record: A mapping of field name to field value for the new record.
        :type record: dict
        :returns: The raw JSON response body for the created record.
        :rtype: str
        :raises HTTPException: If the Preservica API returns an unexpected HTTP error status.
        """
        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




    def record(self, reference: str) -> dict:
        """
        Return a single record by its unique reference.

        :param reference: The unique reference of the record to retrieve.
        :type reference: str
        :returns: The record as a JSON dictionary.
        :rtype: dict
        :raises HTTPException: If the Preservica API returns an unexpected HTTP error status.
        """
        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




    def records(self, table: Table) -> List[dict]:
        """
        Return all records from an authority table.

        :param table: The authority table whose records should be retrieved.
        :type table: Table
        :returns: A list of records, each represented as a JSON dictionary with field values
            expanded.
        :rtype: list[dict]
        :raises HTTPException: If the Preservica API returns an unexpected HTTP error status.
        """
        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




    def add_table(self, new_table: Table):
        """
        Create a new authority (reference metadata) table in Preservica.

        :param new_table: The Table object describing the table to create.  Set
            ``new_table.description``, ``new_table.displayField``,
            ``new_table.metadataConnections``, and ``new_table.fields`` before calling this
            method if those optional attributes are required.
        :type new_table: Table
        :returns: The newly created authority table, fetched fresh from the API so that the
            ``reference`` attribute is populated.
        :rtype: Table
        :raises HTTPException: If the Preservica API returns an unexpected HTTP error status.
        """
        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




    def table(self, reference: str) -> Table:
        """
        Fetch an authority table by its unique reference.

        :param reference: The unique reference identifier for the authority table.
        :type reference: str
        :returns: The authority table with all available attributes populated.
        :rtype: Table
        :raises HTTPException: If the Preservica API returns an unexpected HTTP error status.
        """
        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




    def tables(self) -> Set[Table]:
        """
        Return all reference metadata (authority) tables in the tenancy.

        :returns: A set of all authority tables.
        :rtype: set[Table]
        :raises HTTPException: If the Preservica API returns an unexpected HTTP error status.
        """
        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