Skip to content

Instantly share code, notes, and snippets.

@blankdots
Last active April 26, 2021 15:49
Show Gist options
  • Save blankdots/7982a3ecdd1497576a7578b09ec9006c to your computer and use it in GitHub Desktop.
Save blankdots/7982a3ecdd1497576a7578b09ec9006c to your computer and use it in GitHub Desktop.
gitlab_api.py
"""Get information on Gitlab projects
"""
from typing import AsyncGenerator, Dict
import httpx
import logging
import click
import sys
import asyncio
from datetime import datetime
FORMAT = (
"[%(asctime)s][%(name)s][%(process)d %(processName)s][%(levelname)-8s] (L:%(lineno)s) %(funcName)s: %(message)s"
)
logging.basicConfig(format=FORMAT, datefmt="%Y-%m-%d %H:%M:%S")
logging.StreamHandler(sys.stdout)
LOG = logging.getLogger(__name__)
LOG.setLevel(logging.DEBUG)
BASE_URL = "gitlab API"
SESSION = httpx.AsyncClient()
async def _filter_by_user(token: str) -> AsyncGenerator:
"""."""
results: Dict = dict()
never_signed_in = list()
too_old_sign_in = list()
next_page: int = 1
per_page: int = 10
has_more = True
while has_more:
r = await SESSION.get(
f"{BASE_URL}/users",
params={
"private_token": token,
"external": "true",
"order_by": "id",
"per_page": per_page,
"page": next_page,
},
)
if r.status_code == 200:
total_pages = int(r.headers["X-Total-Pages"])
data = r.json()
for user in data:
if user["last_sign_in_at"]:
delta = datetime.now() - datetime.strptime(user["last_sign_in_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
if delta.days > 180:
too_old_sign_in.append(user["username"])
else:
never_signed_in.append(user["username"])
if r.headers["X-Next-Page"] and total_pages >= next_page:
next_page = int(r.headers["X-Next-Page"])
else:
has_more = False
else:
LOG.error(f"Error retrieving data. API call returned a {r.status_code}")
has_more = False
results["oldSignIn"] = too_old_sign_in
results["neverSignedIn"] = never_signed_in
yield results
async def _filter_by_project(token: str) -> AsyncGenerator:
"""."""
results: Dict = dict()
no_activity = list()
too_old_activity = list()
empty_repo = list()
next_page: int = 1
per_page: int = 10
has_more = True
while has_more:
r = await SESSION.get(
f"{BASE_URL}/projects",
params={
"private_token": token,
"order_by": "id",
"per_page": per_page,
"page": next_page,
},
)
if r.status_code == 200:
total_pages = int(r.headers["X-Total-Pages"])
data = r.json()
for project in data:
if project["last_activity_at"]:
delta = datetime.now() - datetime.strptime(project["last_activity_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
if delta.days > 180:
too_old_activity.append(
{
"url": project["web_url"],
"forks": project["forks_count"],
"visibility": project["visibility"],
}
)
else:
no_activity.append(
{
"url": project["web_url"],
"forks": project["forks_count"],
"visibility": project["visibility"],
}
)
if "empty_repo" in project and project["empty_repo"]:
empty_repo.append({"url": project["web_url"]})
if r.headers["X-Next-Page"] and total_pages >= next_page:
next_page = int(r.headers["X-Next-Page"])
else:
has_more = False
else:
LOG.error(f"Error retrieving data. API call returned a {r.status_code}")
has_more = False
results["oldActivity"] = too_old_activity
results["noActivity"] = no_activity
results["emptyRepo"] = empty_repo
yield results
async def print_users(token: str) -> None:
"""."""
users = [res async for res in _filter_by_user(token)]
print(users)
async def print_projects(token: str) -> None:
"""."""
projects = [res async for res in _filter_by_project(token)]
print(projects)
async def main(token: str) -> None:
"""."""
await print_users(token)
await print_projects(token)
await SESSION.aclose()
@click.command()
@click.option("-t", "--token", help="Token for the API.")
def cli(token: str) -> None:
"""."""
LOG.info(f"Start ==== >")
asyncio.run(main(token))
LOG.info(f"< ==== End")
if __name__ == "__main__":
cli()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment