OpenDMA REST client for Python

Please see OpenDMA API for Python for an introduction to OpenDMA and it's Python API.

Provides an OpenDMA Adaptor consuming the OpenDMA REST API.

Installation

This project is available on PyPI. You can simply install it with pip:

pip install opendma-remote

Usage

Import this module and use the connect function to establish an OdmaSession with a REST-ful OpenDMA service:

import opendma.remote

session = opendma.remote.connect(endpoint="http://127.0.0.1:8080/opendma/",
                                 username="user",
								 password="secret")

Example

Run the tutorial REST service docker container:

docker run -p 8080:8080 ghcr.io/opendma/tutorial-xmlrepo:0.8

It will provide the tutorial xml repository. Make sure that this service is available by opening
http://localhost:8080/opendma
in a web browser.

docker run -p 8080:8080 opendma/tutorial-xmlrepo:latest

docker run -p 8090:8080 ghcr.io/opendma/tutorial-xmlrepo:latest

docker run -d --name opendma-tutorial-xmlrepo -p 8080:8080 ghcr.io/opendma/tutorial-xmlrepo:latest

The following python application will establish a connection to this service and print out the entire repository:

from typing import Optional
from opendma.api import OdmaType, OdmaQName, OdmaId, OdmaObject, OdmaClass, OdmaPropertyInfo, OdmaFolder, OdmaAssociation, OdmaDocument, OdmaContentElement, OdmaDataContentElement, OdmaContent, OdmaVersionCollection, OdmaServiceException
import opendma.remote


def printClassTree(clazz: OdmaClass, indent: int = 0) -> None:
    print(f"{'  ' * indent}{clazz.get_qname()}")
    aspects = clazz.get_aspects()
    if aspects is not None:
        for aspect in aspects:
            print(f"{'  ' * indent}  @{aspect.get_qname()}")
    subclasses = clazz.get_sub_classes()
    if subclasses is not None:
        for subclass in subclasses:
            printClassTree(subclass, indent + 1)


def printObjectInfo(obj: OdmaObject, indent: int, include_props: bool, expand_special: bool):
    print(f"{'  ' * indent}{obj.get_id()} {obj.get_odma_class().get_qname()}")
    if not include_props:
        return
    properties = obj.get_odma_class().get_properties()
    indent = indent + 1
    for propinfo in properties:
        pi: OdmaPropertyInfo = propinfo
        prop = obj.get_property(pi.get_qname())
        type = prop.get_type()
        value = prop.get_value()
        if prop.is_multi_value():
            if type == OdmaType.REFERENCE:
                print(f"{'  ' * indent}{pi.get_qname()}=")
                for pos, item in enumerate(value):
                    id = item.get_id()
                    if '/' in str(id):
                        printObjectInfo(item, indent+1, True, False)
                    else:
                        print(f"{'  ' * (indent+3)}* Object<{id}>")
            else:
                print(f"{'  ' * indent}{pi.get_qname()}=")
                for pos, item in enumerate(value):
                    print(f"{'  ' * (indent+3)}[{pos}] {item}")
        else:
            if value == None:
                print(f"{'  ' * indent}{pi.get_qname()}=None")
            else:
                if type == OdmaType.REFERENCE:
                    if expand_special and (isinstance(value, OdmaVersionCollection) or isinstance(value, OdmaContentElement)):
                        print(f"{'  ' * indent}{pi.get_qname()}=")
                        printObjectInfo(value, indent+1, True, False)
                    else:
                        print(f"{'  ' * indent}{pi.get_qname()}=Object<{value.get_id()}>")
                elif type == OdmaType.CONTENT:
                    cnt: OdmaContent =  value
                    stream = cnt.get_stream()
                    data = stream.read(25)
                    hex_bytes = ' '.join(f'{b:02x}' for b in data)
                    print(f"{'  ' * indent}{pi.get_qname()}= {cnt.get_size()} bytes; first 25 bytes: {hex_bytes}")
                else:
                    print(f"{'  ' * indent}{pi.get_qname()}={value}")


def printFolderTree(folder: OdmaFolder, indent: int = 0) -> None:
    print(f"{'  ' * indent}{folder.get_title()}")
    associations = folder.get_associations()
    if associations is not None:
        for association in associations:
            assoc: OdmaAssociation = association
            print(f"{'  ' * indent}  -{assoc.get_name()}")
            printObjectInfo(assoc.get_containable(), indent + 2, True, True)
    subfolders = folder.get_sub_folders()
    if subfolders is not None:
        for subfolder in subfolders:
            printFolderTree(subfolder, indent + 1)


def printRepo(endpoint: str, username: Optional[str] = None, password: Optional[str] = None, includeClassTree: bool = True, includeFolderTree: bool = True) -> None:
    print(f"Trying to connect...")
    session = opendma.remote.connect(endpoint=endpoint, username=username, password=password, requestTraceLevel = 0)
    print(f"Connected.")
    print(f"Repositories: {session.get_repository_ids()}")
    repoId = session.get_repository_ids()[0]
    print(f"Trying to get repository {repoId}...")
    repo = session.get_repository(repoId)
    print(f"Got repository.")
    print(f"  Id: {repo.get_id()}")
    print(f"  Display Name: {repo.get_display_name()}")
    if includeClassTree:
        print(f"")
        print(f"Class tree:")
        rootClass = repo.get_root_class()
        printClassTree(rootClass)
    if includeFolderTree:
        print(f"")
        print(f"Folder tree:")
        rootFolder = repo.get_root_folder()
        if rootFolder is None:
            print(f"No root folder is in this repository")
        else:
            printFolderTree(rootFolder)


if __name__ == "__main__":
    printRepo(endpoint="http://127.0.0.1:8080/opendma/")