added snippet interface and httpclient
This commit is contained in:
101
scilog/scilog.py
101
scilog/scilog.py
@ -1,41 +1,90 @@
|
||||
from .authclient import AuthClient, AuthError, AUTH_HEADERS
|
||||
from .utils import post_request, get_request
|
||||
from __future__ import annotations
|
||||
|
||||
from .authclient import AuthMixin, AuthError, HEADER_JSON
|
||||
from .mkfilt import make_filter
|
||||
from .httpclient import HttpClient
|
||||
from .snippet import Snippet
|
||||
from typing import TypeVar, Union, List, Type, get_type_hints
|
||||
import functools
|
||||
|
||||
|
||||
class SciLog(AuthClient):
|
||||
class Basesnippet(Snippet):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_properties(
|
||||
id=str,
|
||||
parentId=str,
|
||||
ownerGroup=str,
|
||||
accessGroups=list,
|
||||
snippetType=str,
|
||||
isPrivate=bool,
|
||||
createdAt=str,
|
||||
createdBy=str,
|
||||
updatedAt=str,
|
||||
updateBy=str,
|
||||
subsnippets=List[type(Basesnippet)],
|
||||
tags=List[str],
|
||||
dashboardName=str,
|
||||
files=str,
|
||||
location=str,
|
||||
defaultOrder=int,
|
||||
linkType=str,
|
||||
versionable=bool,
|
||||
deleted=bool)
|
||||
|
||||
|
||||
def authenticate(self, username, password):
|
||||
url = self.address + "/users/login"
|
||||
auth_payload = {
|
||||
"principal": username,
|
||||
"password": password
|
||||
}
|
||||
res = post_request(url, auth_payload, AUTH_HEADERS)
|
||||
try:
|
||||
token = "Bearer " + res["token"]
|
||||
except KeyError as e:
|
||||
raise SciLogAuthError(res) from e
|
||||
else:
|
||||
return token
|
||||
|
||||
class Paragraph(Basesnippet):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.set_properties(textcontent=str, isMessage=str)
|
||||
|
||||
|
||||
class SciLogRestAPI(HttpClient):
|
||||
def __init__(self, url):
|
||||
super().__init__(url)
|
||||
self._verify_certificate = False
|
||||
|
||||
|
||||
class SciLog():
|
||||
|
||||
def __init__(self, url="https://lnode2.psi.ch/api/v1"):
|
||||
self.http_client = SciLogRestAPI(url)
|
||||
self.logbook_id = None
|
||||
self.owner_group = None
|
||||
|
||||
def select_logbook(self, logbook:type(Basesnippet)):
|
||||
self.logbook_id = logbook.id
|
||||
self.owner_group = logbook.ownerGroup
|
||||
|
||||
def get_snippets(self, **kwargs):
|
||||
url = self.address + "/basesnippets"
|
||||
params = make_filter(**kwargs)
|
||||
headers = self.auth_headers
|
||||
return get_request(url, params=params, headers=headers)
|
||||
url = self.http_client.address + "/basesnippets"
|
||||
params = self.http_client.make_filter(where=kwargs)
|
||||
headers = HEADER_JSON.copy()
|
||||
return self.http_client.get_request(url, params=params, headers=headers)
|
||||
|
||||
def send_message(self, msg, **kwargs):
|
||||
url = self.http_client.address + "/basesnippets"
|
||||
payload = kwargs
|
||||
kwargs["textcontent"] = msg
|
||||
headers = HEADER_JSON.copy()
|
||||
return self.http_client.post_request(url, payload=payload, headers=headers)
|
||||
|
||||
def post_snippet(self, **kwargs):
|
||||
url = self.address + "/basesnippets"
|
||||
url = self.http_client.address + "/basesnippets"
|
||||
payload = kwargs
|
||||
headers = self.auth_headers
|
||||
return post_request(url, payload=payload, headers=headers)
|
||||
headers = HEADER_JSON.copy()
|
||||
return self.http_client.post_request(url, payload=payload, headers=headers)
|
||||
|
||||
def get_logbooks(self, **kwargs):
|
||||
url = self.http_client.address + "/basesnippets"
|
||||
snippet = Basesnippet()
|
||||
snippet.import_dict(kwargs)
|
||||
snippet.snippetType = "logbook"
|
||||
params = self.http_client.make_filter(where=snippet.to_dict(include_none=False))
|
||||
headers = HEADER_JSON.copy()
|
||||
return Basesnippet.from_http_response(self.http_client.get_request(url, params=params, headers=headers))
|
||||
|
||||
|
||||
class SciLogAuthError(AuthError):
|
||||
pass
|
||||
|
||||
|
||||
|
||||
|
Reference in New Issue
Block a user