Skip to content
Snippets Groups Projects
Commit 25047f6c authored by GitLab Runner 1149's avatar GitLab Runner 1149
Browse files

Automated Save by CI Run

parent e38e6394
No related branches found
No related tags found
No related merge requests found
......@@ -29970,7 +29970,7 @@ paths:
content:
application/json:
schema:
$ref: https://netvs-devel.scc.kit.edu/netdb_4.0_ta.yml
$ref: https://netvs-devel.scc.kit.edu/netdb_4_0_ta.yml
required: true
responses:
200:
......
### AUTOGENERATED
import requests
import os
from typing import Union
from pprint import pformat
import logging
logger = logging.getLogger(__name__)
# ExeccutionError is derived from ValueError to keep backward compatibility.
class ExecutionError(ValueError):
......@@ -25,20 +28,35 @@ class ExplicitNull(object):
class APIEndpoint(object):
def __init__(self, base_url: str, token: str, **kwargs):
self.base_url = base_url
def __init__(self, netdb_base_url:str=None, netdb_token:str=None, **kwargs):
self.base_url = netdb_base_url if netdb_base_url is not None kwargs.get('base_url', None)
self.version = '4.0'
self.token = token
self.token = netdb_token if netdb_token is not None kwargs.get('token', None)
class APISession(object):
def __init__(self, endpoint: APIEndpoint):
def __init__(self, endpoint: APIEndpoint, protocol: str='https', update_check: bool='NETDB_CLIENT_DISABLE_UPDATE_CHECK' not in os.environ):
self._build_version = '4.0.701'
self.session = requests.session()
self.session.headers.update({'Authorization': 'Bearer ' + endpoint.token})
self.api_root = f'{protocol}://{endpoint.base_url}'
if 'www-net' in endpoint.base_url:
self.base_url = f'https://{endpoint.base_url}/api/{endpoint.version}'
self.base_url = f'{self.api_root}/api/{endpoint.version}'
else:
self.base_url = f'https://{endpoint.base_url}/{endpoint.version}'
self.base_url = f'{self.api_root}/{endpoint.version}'
try:
if update_check:
versions = requests.get(self.api_root).json()[0]
cur_v = None
ref = [int(r) for r in endpoint.version.split('.')]
for v in versions:
if v['major'] == ref[0] and v['minor'] == ref[1]:
if self._build_version != v['numeric']:
logger.warning(f"This library was built for NETDB API version {self._build_version}, but API reports {v['numeric']}. Please consider updating. "
"(You can disable this message by setting NETDB_CLIENT_DISABLE_UPDATE_CHECK in your environment or by passing update_check=False to APISession)")
break
except:
pass
def execute_ta(self, ta: list, dry_mode: bool = False, dict_mode: bool = False, su_login_name: str = None) -> Union[list, dict]:
res = self.session.post(f'{self.base_url}/wapi/transaction/execute?dry_mode={str(dry_mode).lower()}&dict_mode={str(dict_mode).lower()}{"" if su_login_name is None else "&su_login_name="+su_login_name}', json=ta)
......
......@@ -35944,7 +35944,7 @@ paths:
content:
application/json:
schema:
$ref: https://netvs-devel.scc.kit.edu/netdb_4.1_ta.yml
$ref: https://netvs-devel.scc.kit.edu/netdb_4_1_ta.yml
required: true
responses:
200:
### AUTOGENERATED
import requests
import os
from typing import Union
from pprint import pformat
import logging
logger = logging.getLogger(__name__)
# ExeccutionError is derived from ValueError to keep backward compatibility.
class ExecutionError(ValueError):
......@@ -25,20 +28,35 @@ class ExplicitNull(object):
class APIEndpoint(object):
def __init__(self, base_url: str, token: str, **kwargs):
self.base_url = base_url
def __init__(self, netdb_base_url:str=None, netdb_token:str=None, **kwargs):
self.base_url = netdb_base_url if netdb_base_url is not None kwargs.get('base_url', None)
self.version = '4.1'
self.token = token
self.token = netdb_token if netdb_token is not None kwargs.get('token', None)
class APISession(object):
def __init__(self, endpoint: APIEndpoint):
def __init__(self, endpoint: APIEndpoint, protocol: str='https', update_check: bool='NETDB_CLIENT_DISABLE_UPDATE_CHECK' not in os.environ):
self._build_version = '4.1.159'
self.session = requests.session()
self.session.headers.update({'Authorization': 'Bearer ' + endpoint.token})
self.api_root = f'{protocol}://{endpoint.base_url}'
if 'www-net' in endpoint.base_url:
self.base_url = f'https://{endpoint.base_url}/api/{endpoint.version}'
self.base_url = f'{self.api_root}/api/{endpoint.version}'
else:
self.base_url = f'https://{endpoint.base_url}/{endpoint.version}'
self.base_url = f'{self.api_root}/{endpoint.version}'
try:
if update_check:
versions = requests.get(self.api_root).json()[0]
cur_v = None
ref = [int(r) for r in endpoint.version.split('.')]
for v in versions:
if v['major'] == ref[0] and v['minor'] == ref[1]:
if self._build_version != v['numeric']:
logger.warning(f"This library was built for NETDB API version {self._build_version}, but API reports {v['numeric']}. Please consider updating. "
"(You can disable this message by setting NETDB_CLIENT_DISABLE_UPDATE_CHECK in your environment or by passing update_check=False to APISession)")
break
except:
pass
def execute_ta(self, ta: list, dry_mode: bool = False, dict_mode: bool = False, su_login_name: str = None) -> Union[list, dict]:
res = self.session.post(f'{self.base_url}/wapi/transaction/execute?dry_mode={str(dry_mode).lower()}&dict_mode={str(dict_mode).lower()}{"" if su_login_name is None else "&su_login_name="+su_login_name}', json=ta)
......
......@@ -33807,7 +33807,7 @@ paths:
content:
application/json:
schema:
$ref: https://netvs-devel.scc.kit.edu/netdb_4.2_ta.yml
$ref: https://netvs-devel.scc.kit.edu/netdb_4_2_ta.yml
required: true
responses:
200:
### AUTOGENERATED
import requests
import os
from typing import Union
from pprint import pformat
import logging
logger = logging.getLogger(__name__)
# ExeccutionError is derived from ValueError to keep backward compatibility.
class ExecutionError(ValueError):
......@@ -25,20 +28,35 @@ class ExplicitNull(object):
class APIEndpoint(object):
def __init__(self, base_url: str, token: str, **kwargs):
self.base_url = base_url
def __init__(self, netdb_base_url:str=None, netdb_token:str=None, **kwargs):
self.base_url = netdb_base_url if netdb_base_url is not None kwargs.get('base_url', None)
self.version = '4.2'
self.token = token
self.token = netdb_token if netdb_token is not None kwargs.get('token', None)
class APISession(object):
def __init__(self, endpoint: APIEndpoint):
def __init__(self, endpoint: APIEndpoint, protocol: str='https', update_check: bool='NETDB_CLIENT_DISABLE_UPDATE_CHECK' not in os.environ):
self._build_version = '4.2.14'
self.session = requests.session()
self.session.headers.update({'Authorization': 'Bearer ' + endpoint.token})
self.api_root = f'{protocol}://{endpoint.base_url}'
if 'www-net' in endpoint.base_url:
self.base_url = f'https://{endpoint.base_url}/api/{endpoint.version}'
self.base_url = f'{self.api_root}/api/{endpoint.version}'
else:
self.base_url = f'https://{endpoint.base_url}/{endpoint.version}'
self.base_url = f'{self.api_root}/{endpoint.version}'
try:
if update_check:
versions = requests.get(self.api_root).json()[0]
cur_v = None
ref = [int(r) for r in endpoint.version.split('.')]
for v in versions:
if v['major'] == ref[0] and v['minor'] == ref[1]:
if self._build_version != v['numeric']:
logger.warning(f"This library was built for NETDB API version {self._build_version}, but API reports {v['numeric']}. Please consider updating. "
"(You can disable this message by setting NETDB_CLIENT_DISABLE_UPDATE_CHECK in your environment or by passing update_check=False to APISession)")
break
except:
pass
def execute_ta(self, ta: list, dry_mode: bool = False, dict_mode: bool = False, su_login_name: str = None) -> Union[list, dict]:
res = self.session.post(f'{self.base_url}/wapi/transaction/execute?dry_mode={str(dry_mode).lower()}&dict_mode={str(dict_mode).lower()}{"" if su_login_name is None else "&su_login_name="+su_login_name}', json=ta)
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment