gitlab_api.py

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()