Merge pull request #21 from BITNP:swo/use-httpx

refactor: use httpx as http client in place of requests

bitsrun/user.py

@@ -4,7 +4,7 @@
 from hashlib import sha1
 from typing import Dict, Literal, Optional, TypedDict, Union
 
-from requests import Session
+import httpx
 
 from bitsrun.utils import fkbase64, parse_homepage, xencode
 
@@ -35,7 +35,7 @@ def __init__(self, username: str, password: str):
         self.password = password
 
         self.ip, self.acid = parse_homepage(api_base=_API_BASE)
-        self.session = Session()
+        self.client = httpx.Client(base_url=_API_BASE)
 
     def login(self) -> UserResponseType:
         logged_in_user = self._user_validate()
@@ -57,7 +57,7 @@ def logout(self) -> UserResponseType:
 
     def _do_action(self, action: Action) -> UserResponseType:
         params = self._make_params(action)
-        response = self.session.get(_API_BASE + "/cgi-bin/srun_portal", params=params)
+        response = self.client.get("/cgi-bin/srun_portal", params=params)
         return json.loads(response.text[6:-1])
 
     def _get_user_info(self) -> Optional[str]:
@@ -67,7 +67,7 @@ def _get_user_info(self) -> Optional[str]:
             The username of the current logged in user if exists.
         """
 
-        resp = self.session.get(_API_BASE + "/cgi-bin/rad_user_info")
+        resp = self.client.get("/cgi-bin/rad_user_info")
         data = resp.text
 
         if data == "not_online_error":
@@ -99,7 +99,7 @@ def _user_validate(self) -> Optional[str]:
 
     def _get_token(self) -> str:
         params = {"callback": "jsonp", "username": self.username, "ip": self.ip}
-        response = self.session.get(_API_BASE + "/cgi-bin/get_challenge", params=params)
+        response = self.client.get("/cgi-bin/get_challenge", params=params)
         result = json.loads(response.text[6:-1])
         return result["challenge"]
 

bitsrun/utils.py

@@ -2,9 +2,8 @@
 from base64 import b64encode
 from html.parser import HTMLParser
 from typing import Tuple
-from urllib.parse import parse_qs, urlparse
 
-import requests
+import httpx
 
 
 def parse_homepage(api_base: str) -> Tuple[str, str]:
@@ -18,11 +17,10 @@ def parse_homepage(api_base: str) -> Tuple[str, str]:
         A tuple of (ip, acid) of the current session.
     """
 
-    res = requests.get(api_base)
+    res = httpx.get(api_base, follow_redirects=True)
 
     # ac_id appears in the url query parameter of the redirected URL
-    query = parse_qs(urlparse(res.url).query)
-    ac_id = query.get("ac_id")
+    ac_id = res.url.params.get("ac_id")
 
     if not ac_id:
         raise Exception("failed to get acid")