diff --git a/eph_client/client.py b/eph_client/client.py
index 7af67625a2b7488a8236ece42797db71bbc1874f..609ef772dabf1405cc95014b92af3856cf094b31 100644
--- a/eph_client/client.py
+++ b/eph_client/client.py
@@ -1,5 +1,6 @@
from datetime import datetime
import logging
+import re
import requests
import typing
import zeep
@@ -32,11 +33,18 @@ class EphClient:
config: typing.Optional[models.EphSoapService]
) -> typing.Tuple[typing.Optional[zeep.client.Client],
typing.Optional[zeep.proxy.ServiceProxy]]:
+
if not config:
return None, None
- client = zeep.Client(config.wsdl,
- transport=zeep.transports.Transport(session=requests.Session()))
- client.transport.session.headers.update(config.headers)
+
+ session = requests.Session()
+ session.headers.update(config.headers)
+ transport = zeep.transports.Transport(session=session)
+
+ settings = zeep.Settings(extra_http_headers=config.headers)
+ client = zeep.Client(
+ settings=settings, wsdl=config.wsdl, transport=transport)
+
return client, client.service
def upload_file(self, doc: models.Doc) -> str:
@@ -71,18 +79,23 @@ class EphClient:
ref_sak, sak = None, None
if req.sak_id is None:
- ref_sak, sak = self.create_sak_obj(do_factory, model_factory, off_tittel, tittel)
- prim_klassering, sek_klassering = self.create_klassering(model_factory, ref_sak, req)
+ ref_sak, sak = self.create_sak_obj(
+ do_factory, model_factory, off_tittel, tittel)
+ prim_klassering, sek_klassering = self.create_klassering(
+ model_factory, ref_sak, req)
ins.extend([sak, prim_klassering, sek_klassering])
else:
sak = self.fetch_by_id("sak", req.sak_id)
if sak is None:
- ClientError("Could not find sak, invalid sak_id " + str(req.sak_id))
+ ClientError(
+ "Could not find sak, invalid sak_id " + str(req.sak_id))
- journalpost, ref_jp = self.create_journalpost(do_factory, model_factory, ref_sak, req)
+ journalpost, ref_jp = self.create_journalpost(
+ do_factory, model_factory, ref_sak, req)
ins.append(journalpost)
- avsender_mottaker = self.create_avsender_mottaker(model_factory, ref_jp, req)
+ avsender_mottaker = self.create_avsender_mottaker(
+ model_factory, ref_jp, req)
ins.append(avsender_mottaker)
data_obj = self.create_document(do_factory, model_factory, ref_jp, req)
@@ -190,7 +203,8 @@ class EphClient:
if tittel is None or person_navn is None:
raise ClientError("Missing required fields tittel or person_navn'")
elif person_navn in tittel:
- raise ClientError("Invalid format in tittel, tittel should not include person_navn")
+ raise ClientError(
+ "Invalid format in tittel, tittel should not include person_navn")
hide_name = "***** *****"
off_tittel = "{} - {}".format(hide_name, tittel)
@@ -203,12 +217,36 @@ class EphClient:
try:
insert_resp = self.object_service.Insert(dataObjects=af_data_objects,
- identity=
- self.config.object_service.identity.dict())
+ identity=self.config.object_service.identity.dict())
except zeep.exceptions.Fault as ex:
raise ClientError("Create sak failed, " + ex.message)
return insert_resp
+ def search_admenheter_by_kortnavn(self, kortnavn: str) -> typing.List[models.EphAdministrativEnhet]:
+ badchars = re.findall(r'[^\w-]', kortnavn)
+ if badchars:
+ raise ClientError(
+ f'"{kortnavn}" contains illegal characters: {badchars}')
+
+ type_factory = self.object_client.type_factory("ns1")
+
+ query_arg = type_factory.FilteredQueryArguments(
+ DataObjectName='AdministrativEnhet',
+ FilterExpression=f"kortnavn = '{kortnavn}'",
+ ReturnTotalCount=True,
+ )
+
+ try:
+ search_resp = self.object_service.FilteredQuery(
+ arguments=query_arg,
+ identity=self.config.document_service.identity.dict())
+ except zeep.exceptions.Fault as e:
+ raise ClientError(f'Search failed, {e.message}')
+
+ if not search_resp.DataObjects:
+ return []
+ return [models.EphAdministrativEnhet.from_orm(x) for x in search_resp.DataObjects.DataObject]
+
def search_sak(self, name: str, emp_nr: int) -> [int]:
"""Sak search by name, emp_nr optional """
if not self.object_client and not self.object_service:
@@ -309,12 +347,12 @@ class EphClient:
return data_obj
- def fetch_by_id(self, db_obj_name, sak_id):
+ def fetch_by_id(self, db_obj_name, id):
"""Fetches objects from EiS"""
pk = {
"Key": "Id",
- "Value": str(sak_id)
+ "Value": str(id)
}
do_factory = self.object_client.type_factory("ns1")
pk_col = do_factory.PrimaryKeyCollection(PrimaryKey=[pk])
@@ -330,9 +368,22 @@ class EphClient:
identity=self.config.object_service.identity.dict())
result = fetch_resp.DataObject
except zeep.exceptions.Fault as ex:
- raise ClientError("Create sak failed, " + ex.message)
+ if ex.message == 'Forespurt data finnes ikke i databasen':
+ return None
+ raise ClientError(
+ f"Fetch of {db_obj_name}({id}) failed, {ex.message}")
return result
+ def get_administrativ_enhet_by_id(self, id: str) -> typing.Optional[models.EphAdministrativEnhet]:
+ try:
+ int(id)
+ except ValueError:
+ raise ClientError(f'Invalid id "{id}". Must be integer')
+ orm_obj = self.fetch_by_id('AdministrativEnhet', id)
+ if orm_obj is None:
+ return None
+ return models.EphAdministrativEnhet.from_orm(orm_obj)
+
def get_client(config: models.EphClientConfig):
return EphClient(config)
diff --git a/eph_client/models.py b/eph_client/models.py
index b24e4b23e2c83fa37448927e31d3b0ddaaa1f9aa..97196197f7adfb806b388fae0e6ee6fa99a0cb88 100644
--- a/eph_client/models.py
+++ b/eph_client/models.py
@@ -7,12 +7,6 @@ import typing
from pydantic import root_validator
-def to_lower_camel(s: str) -> str:
- """Alias generator to avoid breaking PEP8"""
- first, *others = s.split('_')
- return ''.join([first.lower(), *map(str.capitalize, others)])
-
-
def to_upper_camel(s: str) -> str:
"""Alias generator to avoid breaking PEP8"""
parts = s.split('_')
@@ -59,7 +53,7 @@ class EphUploadHeader(BaseModel):
class EphSoapCredentials(BaseModel):
database: str
user_name: str
- password: typing.Optional[str]
+ password: str
external_system_name: str
class Config:
@@ -92,7 +86,8 @@ class Search(BaseModel):
val = values.get(field)
match = re.match(r"^\w+$", val)
if match is None:
- raise ValueError("Invalid config values, cant use in search: " + val)
+ raise ValueError(
+ "Invalid config values, cant use in search: " + val)
return values
@@ -280,4 +275,95 @@ class CreateSakResp(BaseModel):
sek_klassering: typing.Optional[EphKlassering]
-EphKlassering.update_forward_refs()
+class EphEnhetstype(BaseModel):
+ betegnelse: typing.Optional[str]
+ id: typing.Optional[str]
+ ledetekst: typing.Optional[str]
+ underlagt_enhetstype: typing.Optional['EphEnhetstype']
+ underlagt_enhetstype_id: typing.Optional[str]
+
+ class Config:
+ alias_generator = to_upper_camel
+ allow_population_by_field_name = True
+ orm_mode = True
+
+
+EphEnhetstype.update_forward_refs()
+
+
+class EphJournalstatus(BaseModel):
+ benyttes_av_arkivet: typing.Optional[bool]
+ benyttes_av_leder: typing.Optional[bool]
+ benyttes_av_saksbehandler: typing.Optional[bool]
+ betegnelse: typing.Optional[str]
+ dokumentkontroll: typing.Optional[str]
+ for_eksternt_produserte_dokument: typing.Optional[bool]
+ for_internt_produserte_dokument: typing.Optional[bool]
+ id: typing.Optional[str]
+ kontroll_av_ekspedering: typing.Optional[bool]
+ registreringsansvar: typing.Optional[str]
+ hoved_id: typing.Optional[int]
+ kortkode: typing.Optional[str]
+ standard: typing.Optional[bool]
+
+ class Config:
+ alias_generator = to_upper_camel
+ allow_population_by_field_name = True
+ orm_mode = True
+
+
+class EphJournalEnhet(BaseModel):
+ avsluttet: typing.Optional[typing.Any]
+ betegnelse: typing.Optional[str]
+ id: typing.Optional[str]
+ lokalisering: typing.Optional[str]
+ publisert: typing.Optional[bool]
+ rapportgruppe: typing.Optional[str]
+ standard_journalstatus_arkivpersonale1: typing.Optional[EphJournalstatus]
+ standard_journalstatus_arkivpersonale1Id: typing.Optional[str]
+ standard_journalstatus_arkivpersonale2: typing.Optional[EphJournalstatus]
+ standard_journalstatus_arkivpersonale2Id: typing.Optional[str]
+ standard_journalstatus_arkivpersonale3: typing.Optional[EphJournalstatus]
+ standard_journalstatus_arkivpersonale3Id: typing.Optional[str]
+ standard_journalstatus_saksbehandler1: typing.Optional[EphJournalstatus]
+ standard_journalstatus_saksbehandler1Id: typing.Optional[str]
+ standard_journalstatus_saksbehandler2: typing.Optional[EphJournalstatus]
+ standard_journalstatus_saksbehandler2Id: typing.Optional[str]
+ standard_journalstatus_saksbehandler3: typing.Optional[EphJournalstatus]
+ standard_journalstatus_saksbehandler3Id: typing.Optional[str]
+ standard_journalstatus_arkivpersonale1Hoved_id: typing.Optional[int]
+ standard_journalstatus_arkivpersonale2Hoved_id: typing.Optional[int]
+ standard_journalstatus_arkivpersonale3Hoved_id: typing.Optional[int]
+ standard_journalstatus_saksbehandler1Hoved_id: typing.Optional[int]
+ standard_journalstatus_saksbehandler2Hoved_id: typing.Optional[int]
+ standard_journalstatus_saksbehandler3Hoved_id: typing.Optional[int]
+
+ class Config:
+ alias_generator = to_upper_camel
+ allow_population_by_field_name = True
+ orm_mode = True
+
+
+class EphAdministrativEnhet(BaseModel):
+ betegnelse: typing.Optional[str]
+ enhetsforkortelse_dette_nivaa: typing.Optional[str]
+ enhetstype: typing.Optional[EphEnhetstype]
+ enhetstype_id: typing.Optional[str]
+ fullstendig_sted: typing.Optional[str]
+ id: typing.Optional[int]
+ journal_enhet: typing.Optional[EphJournalEnhet]
+ journal_enhet_id: typing.Optional[str]
+ kortnavn: typing.Optional[str]
+ nedlagt_dato: typing.Optional[typing.Any]
+ opprettet_dato: typing.Optional[typing.Any]
+ overordnet_enhet: typing.Optional['EphAdministrativEnhet']
+ overordnet_enhet_id: typing.Optional[int]
+ rapportgruppe: typing.Optional[str]
+
+ class Config:
+ alias_generator = to_upper_camel
+ allow_population_by_field_name = True
+ orm_mode = True
+
+
+EphAdministrativEnhet.update_forward_refs()
diff --git a/tests/test_client_intg.py b/tests/test_client_intg.py
index 542f40d5243a83aa04c8a7170b2ec76b1f9527b9..2f4f82f094c15d8b27d32968076d9d133ab263b8 100644
--- a/tests/test_client_intg.py
+++ b/tests/test_client_intg.py
@@ -142,3 +142,27 @@ def m_test_search_wrong_name(service_config: models.EphClientConfig):
resp = client.search_sak("wrong name", None)
assert resp.__len__() == 0
+
+
+def m_test_search_administrativ_enhet(service_config: models.EphClientConfig):
+ client = get_client(service_config)
+ resp = client.search_admenheter_by_kortnavn('1800')
+ assert len(resp) == 1
+
+
+def m_test_search_non_existent_administrativ_enhet(service_config: models.EphClientConfig):
+ client = get_client(service_config)
+ resp = client.search_admenheter_by_kortnavn('ZZZZ')
+ assert len(resp) == 0
+
+
+def m_test_get_administrativ_enhet(service_config: models.EphClientConfig):
+ client = get_client(service_config)
+ resp = client.get_administrativ_enhet_by_id(0)
+ assert resp.id == 0
+
+
+def m_test_get_non_existent_administrativ_enhet(service_config: models.EphClientConfig):
+ client = get_client(service_config)
+ resp = client.get_administrativ_enhet_by_id(9999999999)
+ assert resp is None