Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • it-bott-integrasjoner/setra-client
1 result
Show changes
Commits on Source (2)
"""Client for connecting to SETRA API"""
import logging
import urllib.parse
from datetime import date
......@@ -72,6 +73,7 @@ class SetraEndpoints:
abw_order_complete_url: str = "api/abw_order_complete/",
post_add_abw_order_url: str = "api/add_abw_order/",
abw_order_errors_url: str = "api/abw_order_errors/",
last_fs_batch_url: str = "api/batch_fs_last/",
) -> None:
self.baseurl = url
self.batch_url = batch_url
......@@ -90,6 +92,7 @@ class SetraEndpoints:
self.abw_order_complete_url = abw_order_complete_url
self.post_add_abw_order_url = post_add_abw_order_url
self.abw_order_errors_url = abw_order_errors_url
self.last_fs_batch_url = last_fs_batch_url
""" Get endpoints relative to the SETRA API URL. """
......@@ -214,6 +217,11 @@ class SetraEndpoints:
self.baseurl, "/".join((self.abw_order_errors_url, abw_order_id))
)
def last_fs_batch(self, interface: str) -> str:
return urllib.parse.urljoin(
self.baseurl, "/".join((self.last_fs_batch_url, interface))
)
class SetraClient(object):
default_headers = {
......@@ -265,8 +273,7 @@ class SetraClient(object):
params: Optional[Dict[str, Any]] = None,
return_response: Optional[bool] = True,
**kwargs: Any,
) -> Union[JsonType, str, None]:
...
) -> Union[JsonType, str, None]: ...
@overload
def call(
......@@ -277,8 +284,7 @@ class SetraClient(object):
params: Optional[Dict[str, Any]] = None,
return_response: Optional[bool] = True,
**kwargs: Any,
) -> Union[JsonType, str, requests.models.Response, None]:
...
) -> Union[JsonType, str, requests.models.Response, None]: ...
def call(
self,
......@@ -642,6 +648,13 @@ class SetraClient(object):
response: Any = self.get(url)
return self.object_or_data(AbwOrderErrors, response.json())
def get_last_fs_batch(
self, interface: str
) -> Union[Dict[Any, Any], List[Dict[Any, Any]], OutputBatch, List[OutputBatch]]:
url = self.urls.last_fs_batch(interface)
response: Any = self.get(url)
return self.object_or_data(OutputBatch, response)
def get_client(config_dict: Dict[str, Any]) -> SetraClient:
"""
......
"""Models used by the client"""
import datetime
import json
from enum import Enum
......
"""Version for distribution purposes"""
import os
import pkg_resources
......