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.
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.
Inherited Members
- aiohttp.client.ClientSession
- ATTRS
- ws_connect
- get
- options
- head
- post
- put
- patch
- delete
- close
- closed
- connector
- 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
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
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
Inherited Members
- lernsax.util.client.ApiClient
- pack_responses
- jsonrpc
- login
- refresh_session
- logout
- get_tasks
- get_download_url
- get_board
- add_board_entry
- get_notes
- add_note
- delete_note
- send_email
- get_emails
- read_email
- get_email_folders
- read_quickmessages
- send_quickmessage
- get_quickmessage_history
- group_lernsax_quickmessage_history_by_chat
- aiodav.client.Client
- ROOT
- DEFAULT_HTTP_HEADER
- DEFAULT_REQUESTS
- get_full_path
- list
- free
- create_directory
- is_directory
- info
- unlink
- delete
- move
- copy
- download_iter
- download_to
- download_file
- download_directory
- download
- upload_to
- upload_file
- upload_directory
- upload
- get_property
- set_property
- close
- resource