lernsax.lernsax

LernSucks API Wrapper

 1#!/usr/bin/env python
 2
 3""" LernSucks API Wrapper
 4"""
 5
 6import asyncio
 7from typing import List, Union
 8from aiohttp import ClientSession, BasicAuth, ClientResponse
 9from lernsax.util import ApiClient
10import aiodav
11from importlib.util import find_spec
12from logging import getLogger
13
14from time import asctime
15
16logger = getLogger(__name__)
17
18_ORJSON = find_spec("orjson")
19
20if _ORJSON: import orjson as json
21else: import json
22
23class HttpClient(ClientSession):
24    def __init__(self, *args, **kwargs) -> None:
25        self.api: str = kwargs.pop("api_uri", "https://www.lernsax.de/jsonrpc.php")
26        super().__init__(
27            *args,
28            **kwargs,
29            json_serialize= lambda obj, *args, **kwargs: json.dumps(obj).decode() if _ORJSON else json.dumps(obj)
30        )
31        
32    async def request(self, method: str, *args, **kwargs):
33        response = await super().request(method, self.api, *args, **kwargs)
34        self.log_req(response)
35        return response
36
37    @staticmethod
38    def log_req(response: ClientResponse):
39        logger.debug(
40            f"{response.request_info.method} [{asctime()}] -> {response.url}: {response.status} [{response.content_type}]"\
41            f"Received Headers: {response.headers}"
42            )
43        
44class Client(ApiClient, aiodav.Client):
45    """ Main object for handling LernSax access and responses. """
46
47    def __init__(self, email: str, password: str) -> None:
48        self.email: str = email
49        self.password: str = password
50        self.sid: str = ""
51        self.member_of: List[str] = []
52        self.background: asyncio.Task = asyncio.create_task(self.background_task())
53        
54        self.http: HttpClient = HttpClient()
55        self.dav_session: HttpClient = HttpClient(
56            auth = BasicAuth(self.email, self.password))
57        self.dav: aiodav.Client = aiodav.Client(
58            'https://www.lernsax.de/webdav.php/', login=self.email, password=self.password, session=self.dav_session)
59        
60        #* Copying Functions to this client so you don't need to call self.dav.func, higly bogded but it works
61        for func in dir(self.dav):
62            #* dont overwrite functions that already exist
63            if func not in dir(self):
64                #* copy the function or attr
65                setattr(self, func, getattr(self.dav, func)) 
66
67    async def post(self, json: Union[dict, list]) -> dict:
68        return await (await self.http.request("POST", json=json)).json()
69
70    async def exists(self, *args, **kwargs) -> bool:
71        """
72        Workaroung for LernSax WebDav not passing .exist() checks in aiodav even if the dir exists.
73        """
74        return True
75
76    async def background_task(self) -> None:
77        """
78        background task to run cleanup after shutting down and to continuously refresh the sesion
79        """
80        async def refresher():
81            #! refresh session every 5 minutes
82            while True:
83                if self.sid: print(await self.refresh_session())
84                await asyncio.sleep(60*5)
85        try:
86            await refresher()
87        finally:
88            await self.__cleanup()
89    
90    async def __cleanup(self):
91        if self.sid: await self.logout()
92        if not self.http.closed: await self.http.close()
93        if not self.dav_session.closed: await self.dav_session.close()
class HttpClient(aiohttp.client.ClientSession):
24class HttpClient(ClientSession):
25    def __init__(self, *args, **kwargs) -> None:
26        self.api: str = kwargs.pop("api_uri", "https://www.lernsax.de/jsonrpc.php")
27        super().__init__(
28            *args,
29            **kwargs,
30            json_serialize= lambda obj, *args, **kwargs: json.dumps(obj).decode() if _ORJSON else json.dumps(obj)
31        )
32        
33    async def request(self, method: str, *args, **kwargs):
34        response = await super().request(method, self.api, *args, **kwargs)
35        self.log_req(response)
36        return response
37
38    @staticmethod
39    def log_req(response: ClientResponse):
40        logger.debug(
41            f"{response.request_info.method} [{asctime()}] -> {response.url}: {response.status} [{response.content_type}]"\
42            f"Received Headers: {response.headers}"
43            )

First-class interface for making HTTP requests.

HttpClient(*args, **kwargs)
25    def __init__(self, *args, **kwargs) -> None:
26        self.api: str = kwargs.pop("api_uri", "https://www.lernsax.de/jsonrpc.php")
27        super().__init__(
28            *args,
29            **kwargs,
30            json_serialize= lambda obj, *args, **kwargs: json.dumps(obj).decode() if _ORJSON else json.dumps(obj)
31        )
async def request(self, method: str, *args, **kwargs)
33    async def request(self, method: str, *args, **kwargs):
34        response = await super().request(method, self.api, *args, **kwargs)
35        self.log_req(response)
36        return response

Perform HTTP request.

@staticmethod
def log_req(response: aiohttp.client_reqrep.ClientResponse)
38    @staticmethod
39    def log_req(response: ClientResponse):
40        logger.debug(
41            f"{response.request_info.method} [{asctime()}] -> {response.url}: {response.status} [{response.content_type}]"\
42            f"Received Headers: {response.headers}"
43            )
Inherited Members
aiohttp.client.ClientSession
ATTRS
ws_connect
get
options
head
post
put
patch
delete
close
closed
connector
cookie_jar
version
requote_redirect_url
loop
timeout
headers
skip_auto_headers
auth
json_serialize
connector_owner
raise_for_status
auto_decompress
trust_env
trace_configs
detach
class Client(lernsax.util.client.ApiClient, aiodav.client.Client):
45class Client(ApiClient, aiodav.Client):
46    """ Main object for handling LernSax access and responses. """
47
48    def __init__(self, email: str, password: str) -> None:
49        self.email: str = email
50        self.password: str = password
51        self.sid: str = ""
52        self.member_of: List[str] = []
53        self.background: asyncio.Task = asyncio.create_task(self.background_task())
54        
55        self.http: HttpClient = HttpClient()
56        self.dav_session: HttpClient = HttpClient(
57            auth = BasicAuth(self.email, self.password))
58        self.dav: aiodav.Client = aiodav.Client(
59            'https://www.lernsax.de/webdav.php/', login=self.email, password=self.password, session=self.dav_session)
60        
61        #* Copying Functions to this client so you don't need to call self.dav.func, higly bogded but it works
62        for func in dir(self.dav):
63            #* dont overwrite functions that already exist
64            if func not in dir(self):
65                #* copy the function or attr
66                setattr(self, func, getattr(self.dav, func)) 
67
68    async def post(self, json: Union[dict, list]) -> dict:
69        return await (await self.http.request("POST", json=json)).json()
70
71    async def exists(self, *args, **kwargs) -> bool:
72        """
73        Workaroung for LernSax WebDav not passing .exist() checks in aiodav even if the dir exists.
74        """
75        return True
76
77    async def background_task(self) -> None:
78        """
79        background task to run cleanup after shutting down and to continuously refresh the sesion
80        """
81        async def refresher():
82            #! refresh session every 5 minutes
83            while True:
84                if self.sid: print(await self.refresh_session())
85                await asyncio.sleep(60*5)
86        try:
87            await refresher()
88        finally:
89            await self.__cleanup()
90    
91    async def __cleanup(self):
92        if self.sid: await self.logout()
93        if not self.http.closed: await self.http.close()
94        if not self.dav_session.closed: await self.dav_session.close()

Main object for handling LernSax access and responses.

Client(email: str, password: str)
48    def __init__(self, email: str, password: str) -> None:
49        self.email: str = email
50        self.password: str = password
51        self.sid: str = ""
52        self.member_of: List[str] = []
53        self.background: asyncio.Task = asyncio.create_task(self.background_task())
54        
55        self.http: HttpClient = HttpClient()
56        self.dav_session: HttpClient = HttpClient(
57            auth = BasicAuth(self.email, self.password))
58        self.dav: aiodav.Client = aiodav.Client(
59            'https://www.lernsax.de/webdav.php/', login=self.email, password=self.password, session=self.dav_session)
60        
61        #* Copying Functions to this client so you don't need to call self.dav.func, higly bogded but it works
62        for func in dir(self.dav):
63            #* dont overwrite functions that already exist
64            if func not in dir(self):
65                #* copy the function or attr
66                setattr(self, func, getattr(self.dav, func)) 
async def post(self, json: Union[dict, list]) -> dict:
68    async def post(self, json: Union[dict, list]) -> dict:
69        return await (await self.http.request("POST", json=json)).json()
async def exists(self, *args, **kwargs) -> bool:
71    async def exists(self, *args, **kwargs) -> bool:
72        """
73        Workaroung for LernSax WebDav not passing .exist() checks in aiodav even if the dir exists.
74        """
75        return True

Workaroung for LernSax WebDav not passing .exist() checks in aiodav even if the dir exists.

async def background_task(self) -> None:
77    async def background_task(self) -> None:
78        """
79        background task to run cleanup after shutting down and to continuously refresh the sesion
80        """
81        async def refresher():
82            #! refresh session every 5 minutes
83            while True:
84                if self.sid: print(await self.refresh_session())
85                await asyncio.sleep(60*5)
86        try:
87            await refresher()
88        finally:
89            await self.__cleanup()

background task to run cleanup after shutting down and to continuously refresh the sesion