RGBCube/GitHubWrapper
1
Fork 0
mirror of https://github.com/RGBCube/GitHubWrapper synced 2025-05-18 06:55:09 +00:00
Code Issues Activity Actions

Bare Bones, will add components later

Browse source
This commit is contained in:
VarMonke 2022-03-27 16:55:16 +05:30
parent 3f0326a19e
commit 02ff6bbeed
2 changed files with 18 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

28
Github/http.py
View file

@ -1,9 +1,6 @@
#== http.py ==# #== http.py ==#
import aiohttp import aiohttp
from collections import namedtuple
from datetime import datetime
from types import SimpleNamespace from types import SimpleNamespace
import re import re
@ -13,7 +10,6 @@ from .objects import APIOBJECT
from .urls import * from .urls import *
LINK_PARSING_RE = re.compile(r"<(\S+(\S))>; rel=\"(\S+)\"") LINK_PARSING_RE = re.compile(r"<(\S+(\S))>; rel=\"(\S+)\"")
Rates = namedtuple('Rates', ('remaining', 'used', 'total', 'reset_when', 'last_request'))
# aiohttp request tracking / checking bits # aiohttp request tracking / checking bits
async def on_req_start( async def on_req_start(
@ -22,8 +18,7 @@ async def on_req_start(
params: aiohttp.TraceRequestStartParams params: aiohttp.TraceRequestStartParams
) -> None: ) -> None:
"""Before-request hook to make sure we don't overrun the ratelimit.""" """Before-request hook to make sure we don't overrun the ratelimit."""
#print(repr(session), repr(ctx), repr(params)) print(repr(session), repr(ctx), repr(params))
pass
async def on_req_end( async def on_req_end(
session: aiohttp.ClientSession, session: aiohttp.ClientSession,
@ -31,15 +26,7 @@ async def on_req_end(
params: aiohttp.TraceRequestEndParams params: aiohttp.TraceRequestEndParams
) -> None: ) -> None:
"""After-request hook to adjust remaining requests on this time frame.""" """After-request hook to adjust remaining requests on this time frame."""
headers = params.response.headers print(repr(session), repr(ctx), repr(params))
remaining = headers['X-RateLimit-Remaining']
used = headers['X-RateLimit-Used']
total = headers['X-RateLimit-Limit']
reset_when = datetime.fromtimestamp(int(headers['X-RateLimit-Reset']))
last_req = datetime.utcnow()
session._rates = Rates(remaining, used, total, reset_when, last_req)
trace_config = aiohttp.TraceConfig() trace_config = aiohttp.TraceConfig()
trace_config.on_request_start.append(on_req_start) trace_config.on_request_start.append(on_req_start)
@ -55,7 +42,6 @@ async def make_session(*, headers: dict[str, str], authorization: aiohttp.BasicA
headers=headers, headers=headers,
trace_configs=[trace_config] trace_configs=[trace_config]
) )
session._rates = Rates('', '' , '', '', '')
return session return session
# pagination # pagination
@ -65,7 +51,7 @@ class Paginator:
self.session = session self.session = session
self.response = response self.response = response
self.should_paginate = bool(self.response.headers.get('Link', False)) self.should_paginate = bool(self.response.headers.get('Link', False))
types: dict[str, APIOBJECT] = { types: dict[str, User | ...] = {
'user': User, 'user': User,
} }
self.target_type = types[target_type] self.target_type = types[target_type]
@ -112,3 +98,11 @@ async def get_user(session: aiohttp.ClientSession, username: str) -> GitHubUserD
if result.status == 200: if result.status == 200:
return await result.json() return await result.json()
raise UserNotFound raise UserNotFound
async def get_user_repos(session: aiohttp.ClientSession, username: str) -> list:
"""Returns a user's public repos in JSON format."""
result = await session.get(USER_REPOS_URL.format(username))
if result.status == 200:
return await Paginator(session, result, 'repo').exhaust()
raise UserNotFound

12
Github/main.py
View file

@ -61,11 +61,11 @@ class Github:
"""Fetch a Github user from their username.""" """Fetch a Github user from their username."""
return User(await http.get_user(self.session, username), self.session) return User(await http.get_user(self.session, username), self.session)
async def get_repo(self, repo_name: str) -> 'Repo': #async def get_repo(self, repo_name: str) -> 'Repository':
"""Fetch a Github repository from it's name.""" # """Fetch a Github repository from it's name."""
pass # pass
async def get_org(self, org_name: str) -> 'Org': #async def get_org(self, org_name: str) -> 'Org':
"""Fetch a Github organization from it's name""" # """Fetch a Github organization from it's name"""
pass # pass