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""" """Client for connecting to SETRA API"""
import logging import logging
import urllib.parse import urllib.parse
from datetime import date from datetime import date
...@@ -72,6 +73,7 @@ class SetraEndpoints: ...@@ -72,6 +73,7 @@ class SetraEndpoints:
abw_order_complete_url: str = "api/abw_order_complete/", abw_order_complete_url: str = "api/abw_order_complete/",
post_add_abw_order_url: str = "api/add_abw_order/", post_add_abw_order_url: str = "api/add_abw_order/",
abw_order_errors_url: str = "api/abw_order_errors/", abw_order_errors_url: str = "api/abw_order_errors/",
last_fs_batch_url: str = "api/batch_fs_last/",
) -> None: ) -> None:
self.baseurl = url self.baseurl = url
self.batch_url = batch_url self.batch_url = batch_url
...@@ -90,6 +92,7 @@ class SetraEndpoints: ...@@ -90,6 +92,7 @@ class SetraEndpoints:
self.abw_order_complete_url = abw_order_complete_url self.abw_order_complete_url = abw_order_complete_url
self.post_add_abw_order_url = post_add_abw_order_url self.post_add_abw_order_url = post_add_abw_order_url
self.abw_order_errors_url = abw_order_errors_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. """ """ Get endpoints relative to the SETRA API URL. """
...@@ -214,6 +217,11 @@ class SetraEndpoints: ...@@ -214,6 +217,11 @@ class SetraEndpoints:
self.baseurl, "/".join((self.abw_order_errors_url, abw_order_id)) 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): class SetraClient(object):
default_headers = { default_headers = {
...@@ -265,8 +273,7 @@ class SetraClient(object): ...@@ -265,8 +273,7 @@ class SetraClient(object):
params: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None,
return_response: Optional[bool] = True, return_response: Optional[bool] = True,
**kwargs: Any, **kwargs: Any,
) -> Union[JsonType, str, None]: ) -> Union[JsonType, str, None]: ...
...
@overload @overload
def call( def call(
...@@ -277,8 +284,7 @@ class SetraClient(object): ...@@ -277,8 +284,7 @@ class SetraClient(object):
params: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None,
return_response: Optional[bool] = True, return_response: Optional[bool] = True,
**kwargs: Any, **kwargs: Any,
) -> Union[JsonType, str, requests.models.Response, None]: ) -> Union[JsonType, str, requests.models.Response, None]: ...
...
def call( def call(
self, self,
...@@ -642,6 +648,13 @@ class SetraClient(object): ...@@ -642,6 +648,13 @@ class SetraClient(object):
response: Any = self.get(url) response: Any = self.get(url)
return self.object_or_data(AbwOrderErrors, response.json()) 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: def get_client(config_dict: Dict[str, Any]) -> SetraClient:
""" """
......
"""Models used by the client""" """Models used by the client"""
import datetime import datetime
import json import json
from enum import Enum from enum import Enum
......
"""Version for distribution purposes""" """Version for distribution purposes"""
import os import os
import pkg_resources import pkg_resources
......