Alle Metadatenfelder für ein Dokument holen inkl. Ordner- und Registerfeldern

Im Rahmen eines Projekts habe ich folgenden Code gebaut, welcher die Metadaten eines Dokuments inklusive der Ordner- und Registermetadaten holt. Definitiv noch nicht umgesetzt ist eine Lösung, wenn der gleiche Registertyp mehrfach geschachtelt ist. Dennoch wollte ich dieses Beispiel nicht verpuffen lassen und packe es mal einfach hier rein:

from XmlElement import XmlElement
from ecmind_blue_client import Job, Jobs
from ecmind_blue_client.tcp_pool_client import TcpPoolClient as Client

enaio = Client(connection_string="localhost:4000:100", appname="test", username="root", password="optimal")

def get_metadata(internal_name:str, doc_id: int) -> dict:
    query_xml = XmlElement.from_object(
        "DMSQuery",
        {
            "Archive": {
                "ObjectType": {
                    "@internal_name": internal_name,
                    "Fields": {"@field_schema": "all"},
                    "Conditions": {
                        "ConditionObject": {
                            "@internal_name": internal_name,
                            "FieldCondition": {
                                "@internal_name": "OBJECT_ID",
                                "@system": "1",
                                "Value": str(doc_id),
                            },
                        }
                    },
                    "ParentObjects": {"@parent_schema": "all"},
                }
            }
        },
    )

    job = Job(
        jobname=Jobs.DMS_GETRESULTLIST,
        Flags=0,
        XML=query_xml,
        Encoding="UTF-8",
        RequestType="HOL",
        FieldSchema="ALL",
        DateFormat="%Y-%m-%d",
    )

    job_result = enaio.execute(job)
    assert job_result.return_code == 0, job_result.error_message
    result_xml = XmlElement.from_string(job_result.values["XML"])
    result = {}
    level_name = ""
    for e in result_xml.walk():
        if "internal_name" in e.attributes:
            if "table" in e.attributes:
                result[level_name] = {}
            else:
                result[level_name][e.attributes["internal_name"]] = e.text

    return result

Das sollten wir mal verallgemeinern und in eines der ecmind_blue_client-Module bauen. Die Frage ist, wie weit gehen mit den HOL-Anfragen. Im Moment bin ich aber froh, dass ich das konkrete Problem so lösen konnte und hoffe, das Beispiel hilft vielleicht mal jemand anderen weiter.

Danke für diesen hilfreichen Beitrag @rk!

Hier noch ein Beispiel für eine HOL-Query, falls man nicht alle Metadatenfelder des Ordners zurückeralten möchte, sondern nur bestimmte einzelne Felder (in diesem Beispiel wird das Feld „Year“ vom Ordner für jedes Dokument geholt, das mit der entsprechenden ID innerhalb der letzten 10 Jahre gefunden werden konnte:

def get_docs(id) -> List:
    logging.info(f'get_docs({id})')
    query_xml = X.from_object('DMSQuery', {
          'Archive': {
              'ObjectType': {
                  '@internal_name': 'Doc',
                  '@field_schema': 'DEF',
                  'Fields': {
                      'Field': {'@internal_name': 'OBJECT_ID', '@system': '1'}
                  },
                  'Conditions': {
                      'ConditionObject': [
                          {
                              '@internal_name': 'Folder',
                              'FieldGroup': {
                                  '@operator': 'OR', 
                                  'FieldCondition': [
                                      {'@internal_name': 'ID1', 'Value': {'#': id}},
                                      {'@internal_name': 'ID2', 'Value': {'#': id}},
                                  ]},
                              'FieldCondition': [
                                  {'@internal_name': 'Year', '@operator': '>=', 'Value': {'#': str(datetime.today().year - 10)}},
                              ]
                          },
                          {
                              '@internal_name': 'Doc',
                              'FieldGroup': {'@operator': 'OR', 'FieldGroup': doc_types_and_titles}
                          }
                      ]   
                  },
                  'ParentObjects': {
                      '@parent_schema': 'DEF',
                      'ParentObjectType':{
                          '@internal_name': 'Folder',
                          'Fields': {
                              'Field': {'@internal_name': 'Year'}
                          },
                      }
                  },
              }
          }
      })
      job = Job('dms.GetResultList', Flags=0, Encoding='UTF-8', RequestType='HOL', XML=query_xml)
      result = client.execute(job)
      if result.return_code == 0 and result.values['TotalHits'] > 0:
          result_xml = X.from_string(result.values['XML'])
          type_id = int(result_xml['Archive'][0]['ObjectType'][0]['ObjectList'][0]['Object'][0]['ChildObjects'][0]['ObjectType'][0].attributes['id'])
          result_list = []
          for folder in result_xml['Archive'][0]['ObjectType'][0]['ObjectList']: # Loop over Steuerdossier NP folders
              period = folder['Object'][0]['Fields'][0]['Field'][0].text
              result_list.extend([ {'id': int(row.attributes['id']), 'type': type_id, 'year': year } for row in folder['Object'][0]['ChildObjects'][0]['ObjectType'][0]['ObjectList'][0]['Object']])
          return result_list
      else:
          logging.error(result.error_message)
          return []
1 Like