Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/gogs/lister.py
Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines | ): | ||||
if len(self.credentials) > 0: | if len(self.credentials) > 0: | ||||
cred = random.choice(self.credentials) | cred = random.choice(self.credentials) | ||||
username = cred.get("username") | username = cred.get("username") | ||||
self.api_token = cred["password"] | self.api_token = cred["password"] | ||||
logger.warning( | logger.warning( | ||||
"Using authentication credentials from user %s", username or "???" | "Using authentication credentials from user %s", username or "???" | ||||
) | ) | ||||
else: | else: | ||||
raise ValueError("No credentials or API token provided") | # Raises an error on Gogs, or a warning on Gitea | ||||
self.on_anonymous_mode() | |||||
self.max_page_limit = 2 | self.max_page_limit = 2 | ||||
self.session = requests.Session() | self.session = requests.Session() | ||||
self.session.headers.update( | self.session.headers.update( | ||||
{ | { | ||||
"Accept": "application/json", | "Accept": "application/json", | ||||
"User-Agent": USER_AGENT, | "User-Agent": USER_AGENT, | ||||
"Authorization": f"token {self.api_token}", | |||||
} | } | ||||
) | ) | ||||
if self.api_token: | |||||
self.session.headers["Authorization"] = f"token {self.api_token}" | |||||
def on_anonymous_mode(self): | |||||
raise ValueError("No credentials or API token provided") | |||||
def state_from_dict(self, d: Dict[str, Any]) -> GogsListerState: | def state_from_dict(self, d: Dict[str, Any]) -> GogsListerState: | ||||
return GogsListerState(**d) | return GogsListerState(**d) | ||||
def state_to_dict(self, state: GogsListerState) -> Dict[str, Any]: | def state_to_dict(self, state: GogsListerState) -> Dict[str, Any]: | ||||
return asdict(state) | return asdict(state) | ||||
@throttling_retry(before_sleep=before_sleep_log(logger, logging.WARNING)) | @throttling_retry(before_sleep=before_sleep_log(logger, logging.WARNING)) | ||||
def page_request(self, url, params) -> requests.Response: | def page_request(self, url, params) -> requests.Response: | ||||
▲ Show 20 Lines • Show All 84 Lines • Show Last 20 Lines |