delete many github-trends files, remove relatedProjects, unify gitignore

This commit is contained in:
martin-mfg
2025-09-20 10:01:05 +02:00
parent 1354741004
commit b4e33d2643
154 changed files with 17 additions and 11125 deletions
+15
View File
@@ -0,0 +1,15 @@
backend/.vercel
backend/.env
backend/node_modules
backend/*.lock
backend/coverage
backend/benchmarks
backend/vercel_token
frontend/frontend/.env
# IDE
.idea/
.vscode/*
!.vscode/extensions.json
!.vscode/settings.json
*.code-workspace
-16
View File
@@ -1,16 +0,0 @@
.vercel
.env
node_modules
*.lock
.idea/
coverage
benchmarks
vercel_token
# IDE
.vscode/*
!.vscode/extensions.json
!.vscode/settings.json
*.code-workspace
.vercel
+1 -2
View File
@@ -11,6 +11,5 @@
"source": "/frontend/:match*",
"destination": "https://monorepo-test-frontend-neon.vercel.app/:match*"
}
],
"relatedProjects": ["prj_emQEOHSrnlodZCgZ1kyzVsiiGSQm"]
]
}
-2
View File
@@ -1,2 +0,0 @@
# Auto detect text files and perform LF normalization
* text=auto
-39
View File
@@ -1,39 +0,0 @@
name: CI-Backend
on:
push:
branches: [main]
pull_request:
branches: [main]
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python 3.11
uses: actions/setup-python@v2
with:
python-version: 3.11
- name: Install dependencies
run: |
cd backend
python -m pip install --upgrade pip
pip install -r requirements.txt
- name: Test with unittest
run: |
cd backend
python -m unittest
env:
AUTH_TOKEN: ${{ secrets.AUTH_TOKEN }}
MONGODB_PASSWORD: ${{ secrets.MONGODB_PASSWORD }}
- name: Upload coverage to Coveralls
run: |
cd backend
coverage run --source=src -m unittest
coveralls
env:
AUTH_TOKEN: ${{ secrets.AUTH_TOKEN }}
COVERALLS_REPO_TOKEN: ${{ secrets.COVERALLS_REPO_TOKEN }}
MONGODB_PASSWORD: ${{ secrets.MONGODB_PASSWORD }}
-12
View File
@@ -1,12 +0,0 @@
*.pyc
__pycache__
.vscode
backend/.env
backend/.venv
backend/.coverage
backend/gcloud_key.json
frontend/.env
.DS_Store
-15
View File
@@ -1,15 +0,0 @@
[run]
source = src
omit =
./.venv/*
./tests/*
./models/*
*/__init__.py
[report]
omit =
./.venv/*
./tests/*
./models/*
*/__init__.py
-14
View File
@@ -1,14 +0,0 @@
AUTH_TOKEN=abc123
COVERALLS_REPO_TOKEN=abc123
PROD_OAUTH_CLIENT_ID=abc123
PROD_OAUTH_CLIENT_SECRET=abc123
PROD_OAUTH_REDIRECT_URI=abc123
DEV_OAUTH_CLIENT_ID=abc123
DEV_OAUTH_CLIENT_SECRET=abc123
DEV_OAUTH_REDIRECT_URI=abc123
GOOGLE_APPLICATION_CREDENTIALS=abc123
MONGODB_PASSWORD=abc123
-5
View File
@@ -1,5 +0,0 @@
[flake8]
max-line-length = 88
max-complexity = 100
select = B,C,E,F,W,T
ignore = E203, W503, E501
-34
View File
@@ -1,34 +0,0 @@
# This file specifies files that are *not* uploaded to Google Cloud Platform
# using gcloud. It follows the same syntax as .gitignore, with the addition of
# "#!include" directives (which insert the entries of the given .gitignore-style
# file at that point).
#
# For more information, run:
# $ gcloud topic gcloudignore
#
.gcloudignore
# If you would like to upload your .git directory, .gitignore file or files
# from your .gitignore file, remove the corresponding line
# below:
.git
.gitignore
# Python pycache:
__pycache__/
.venv
.coverage
.coveragerc
.flake8
poetry.lock
pyproject.toml
README.md
# Ignored by the build system
/setup.cfg
# remove irrelevant files/folders
deploy
tests
.env-template
.pre-commit-config.yaml
gcloud_key.json
-11
View File
@@ -1,11 +0,0 @@
repos:
- repo: https://github.com/ambv/black
rev: 21.10b0
hooks:
- id: black
language_version: python3.11
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v2.3.0
hooks:
- id: flake8
args: [--config=./backend/.flake8]
-38
View File
@@ -1,38 +0,0 @@
# Backend
## Installation
```
poetry install
poetry run pre-commit install
```
## Run Locally
Navigate to localhost:8000
```
yarn start
```
## Test with Code Coverage
```
yarn test
```
View coverage with GitHub badge or on coveralls.io
## Build
If a new requirement has been added, make sure to update the requirements.txt
```
yarn set-reqs
```
Then, just commit on the main branch (Google Cloud Run takes care of the rest)
## Adding a Secret
Update cloudbuild.yaml, .env, .env-template, and GCP Cloud Run Trigger Substitution Variables.
-14
View File
@@ -1,14 +0,0 @@
service: default
runtime: python311
entrypoint: gunicorn -w 2 -k uvicorn.workers.UvicornWorker src.main:app
#smallest instance class
instance_class: F1
#prevents creating additional instances
automatic_scaling:
min_instances: 0
max_instances: 1
env_variables:
PROD: true
-21
View File
@@ -1,21 +0,0 @@
steps:
- name: node:10.15.1
entrypoint: npm
args: ["install"]
dir: "backend"
- name: node:10.15.1
entrypoint: npm
args: ["run", "create-env"]
dir: "backend"
env:
- "DEV_OAUTH_CLIENT_ID=${_DEV_OAUTH_CLIENT_ID}"
- "DEV_OAUTH_CLIENT_SECRET=${_DEV_OAUTH_CLIENT_SECRET}"
- "DEV_OAUTH_REDIRECT_URI=${_DEV_OAUTH_REDIRECT_URI}"
- "PROD_OAUTH_CLIENT_ID=${_PROD_OAUTH_CLIENT_ID}"
- "PROD_OAUTH_CLIENT_SECRET=${_PROD_OAUTH_CLIENT_SECRET}"
- "PROD_OAUTH_REDIRECT_URI=${_PROD_OAUTH_REDIRECT_URI}"
- "MONGODB_PASSWORD=${_MONGODB_PASSWORD}"
- "SENTRY_DSN=${_SENTRY_DSN}"
- name: "gcr.io/cloud-builders/gcloud"
args: ["app", "deploy", "--appyaml", "./deploy/app.yaml"]
dir: "backend"
-3
View File
@@ -1,3 +0,0 @@
dispatch:
- url: "*/.*"
service: default
-13
View File
@@ -1,13 +0,0 @@
{
"name": "github-trends",
"version": "0.0.1",
"private": true,
"scripts": {
"gen-lang-map": "poetry run python src/data/github/language_map.py",
"start": "poetry run uvicorn src.main:app --reload --port=8000",
"set-reqs": "poetry lock && poetry export -f requirements.txt --output requirements.txt --without-hashes",
"create-env": "printenv > .env",
"test": "poetry run coverage run --source=src -m unittest -v && poetry run coverage report",
"isort": "poetry run isort . --src-path=./src --multi-line=3 --trailing-comma --line-length=88 --combine-as --ensure-newline-before-comments"
}
}
-1673
View File
File diff suppressed because it is too large Load Diff
-36
View File
@@ -1,36 +0,0 @@
[tool.poetry]
name = "github-trends"
version = "0.1.0"
description = ""
authors = ["Abhijit Gupta <avgupta456@gmail.com>"]
license = "MIT"
[tool.poetry.dependencies]
python = "^3.11"
fastapi = "^0.104.1"
uvicorn = {extras = ["standard"], version = "^0.24.0.post1"}
requests = "^2.31.0"
python-dotenv = "^1.0.0"
motor = "^3.3.1"
aiofiles = "^23.2.1"
aiounittest = "^1.4.2"
coveralls = "^3.3.1"
grpcio = "^1.59.2"
gunicorn = "^21.2.0"
pymongo = {extras = ["srv"], version = "^4.6.0"}
pytz = "^2023.3.post1"
sentry-sdk = "^1.34.0"
svgwrite = "^1.4.3"
[tool.poetry.dev-dependencies]
[tool.poetry.group.dev.dependencies]
black = "^23.11.0"
flake8 = "^6.1.0"
isort = "^5.12.0"
pre-commit = "^3.5.0"
pyinstrument = "^4.6.1"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
-39
View File
@@ -1,39 +0,0 @@
aiofiles==23.2.1
aiounittest==1.4.2
annotated-types==0.6.0
anyio==3.7.1
certifi==2023.11.17
charset-normalizer==3.3.2
click==8.1.7
colorama==0.4.6
coverage==6.5.0
coveralls==3.3.1
dnspython==2.4.2
docopt==0.6.2
fastapi==0.104.1
grpcio==1.59.3
gunicorn==21.2.0
h11==0.14.0
httptools==0.6.1
idna==3.4
motor==3.3.2
packaging==23.2
pydantic-core==2.14.5
pydantic==2.5.2
pymongo==4.6.0
pymongo[srv]==4.6.0
python-dotenv==1.0.0
pytz==2023.3.post1
pyyaml==6.0.1
requests==2.31.0
sentry-sdk==1.36.0
sniffio==1.3.0
starlette==0.27.0
svgwrite==1.4.3
typing-extensions==4.8.0
urllib3==2.1.0
uvicorn[standard]==0.24.0.post1
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0
wrapt==1.16.0
@@ -1,64 +0,0 @@
import asyncio
import os
import sys
from datetime import datetime
from typing import Any
from dotenv import find_dotenv, load_dotenv
load_dotenv(find_dotenv())
# Add the parent directory to the Python path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
# flake8: noqa E402
from src.constants import API_VERSION
from src.data.mongo.main import USER_MONTHS
def get_filters(cutoff_date: datetime) -> Any:
return {
"$or": [
{"month": {"$lte": cutoff_date}},
{"version": {"$ne": API_VERSION}},
],
}
async def count_old_rows(cutoff_date: datetime) -> int:
filters = get_filters(cutoff_date)
num_rows = len(await USER_MONTHS.find(filters).to_list(length=None)) # type: ignore
return num_rows
async def delete_old_rows(cutoff_date: datetime):
filters = get_filters(cutoff_date)
result = await USER_MONTHS.delete_many(filters)
print(f"Deleted {result.deleted_count} rows")
async def main():
# Replace 'your_date_field' with the actual name of your date field
cutoff_date = datetime(2024, 12, 31)
count = await count_old_rows(cutoff_date)
if count == 0:
print("No rows to delete.")
return
print(f"Found {count} rows to delete.")
print()
confirmation = input("Are you sure you want to delete these rows? (yes/no): ")
if confirmation.lower() != "yes":
print("Operation canceled.")
return
print()
await delete_old_rows(cutoff_date)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
-104
View File
@@ -1,104 +0,0 @@
import argparse
import asyncio
import json
import os
import sys
from datetime import datetime
# Add the parent directory to the Python path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
os.environ["LOCAL"] = "True"
# flake8: noqa E402
from src.aggregation.layer0 import get_user_data
from src.processing.user import get_top_languages, get_top_repos
from src.processing.wrapped.package import get_wrapped_data
def parse_args():
parser = argparse.ArgumentParser(description="GitHub Trends Script")
parser.add_argument("--user_id", required=True, help="GitHub user ID", type=str)
parser.add_argument(
"--access_token", required=True, help="GitHub access token", type=str
)
parser.add_argument(
"--start_date",
default="2023-01-01",
help="Start date in YYYY-MM-DD format",
type=str,
)
parser.add_argument(
"--end_date",
default="2023-01-31",
help="End date in YYYY-MM-DD format",
type=str,
)
parser.add_argument(
"--timezone", default="America/New_York", help="Timezone", type=str
)
parser.add_argument(
"--output_dir", default="./", help="Output directory path", type=str
)
return parser.parse_args()
async def main():
args = parse_args()
start_date = datetime.strptime(args.start_date, "%Y-%m-%d")
end_date = datetime.strptime(args.end_date, "%Y-%m-%d")
print("Local script running...")
print("User ID:", args.user_id)
print("Access token:", args.access_token)
print("Start date:", start_date)
print("End date:", end_date)
print("Timezone:", args.timezone)
print("Output directory:", args.output_dir)
print()
raw_output = await get_user_data(
args.user_id, start_date, end_date, args.timezone, args.access_token
)
with open(os.path.join(args.output_dir, "raw.json"), "w") as f:
f.write(raw_output.model_dump_json(indent=2))
langs_output = get_top_languages(
raw_output, loc_metric="changed", include_private=True
)
langs_output = (
[json.loads(x.model_dump_json()) for x in langs_output[0]],
langs_output[1],
)
repos_output = get_top_repos(
raw_output, loc_metric="changed", include_private=True, group="none"
)
repos_output = (
[json.loads(x.model_dump_json()) for x in repos_output[0]],
repos_output[1],
)
with open(os.path.join(args.output_dir, "langs.json"), "w") as f:
f.write(json.dumps(langs_output, indent=2))
with open(os.path.join(args.output_dir, "repos.json"), "w") as f:
f.write(json.dumps(repos_output, indent=2))
wrapped_user = get_wrapped_data(raw_output, 2023)
with open(os.path.join(args.output_dir, "wrapped.json"), "w") as f:
f.write(wrapped_user.model_dump_json(indent=2))
print("Wrote output to", args.output_dir)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
View File
@@ -1,3 +0,0 @@
from src.aggregation.layer0.package import get_user_data
__all__ = ["get_user_data"]
@@ -1,498 +0,0 @@
from collections import defaultdict
from datetime import date, datetime
from typing import Any, Dict, List, Optional, Tuple, Union
import pytz
from src.aggregation.layer0.languages import CommitLanguages, get_commit_languages
from src.constants import (
GRAPHQL_NODE_CHUNK_SIZE,
GRAPHQL_NODE_THREADS,
NODE_QUERIES,
PR_FILES,
REST_NODE_THREADS,
)
from src.data.github.graphql import (
RawCalendar,
RawCommit as GraphQLRawCommit,
RawEventsCommit,
RawEventsEvent,
RawRepo,
get_commits,
get_repo,
get_user_contribution_calendar,
get_user_contribution_events,
)
from src.data.github.rest import (
RawCommit as RESTRawCommit,
RawCommitFile,
get_commit_files,
get_repo_commits,
)
from src.models import UserContributions
from src.utils import date_to_datetime, gather
class ContribsList:
def __init__(self):
self.commits: List[RawEventsCommit] = []
self.issues: List[RawEventsEvent] = []
self.prs: List[RawEventsEvent] = []
self.reviews: List[RawEventsEvent] = []
self.repos: List[RawEventsEvent] = []
def add(self, label: str, event: Union[RawEventsCommit, RawEventsEvent]):
if label == "commit" and isinstance(event, RawEventsCommit):
self.commits.append(event)
elif label == "issue" and isinstance(event, RawEventsEvent):
self.issues.append(event)
elif label == "pr" and isinstance(event, RawEventsEvent):
self.prs.append(event)
elif label == "review" and isinstance(event, RawEventsEvent):
self.reviews.append(event)
elif label == "repo" and isinstance(event, RawEventsEvent):
self.repos.append(event)
def get_user_all_contribution_events(
user_id: str,
start_date: datetime,
end_date: datetime,
access_token: Optional[str] = None,
) -> Dict[str, ContribsList]:
repo_contribs: Dict[str, ContribsList] = defaultdict(lambda: ContribsList())
after: Optional[str] = ""
cont = True
while cont:
after_str = after if isinstance(after, str) else ""
response = get_user_contribution_events(
user_id=user_id,
start_date=start_date,
end_date=end_date,
after=after_str,
access_token=access_token,
)
cont = False
node_lists = [
("commit", response.commit_contribs_by_repo),
("issue", response.issue_contribs_by_repo),
("pr", response.pr_contribs_by_repo),
("review", response.review_contribs_by_repo),
]
for event_type, event_list in node_lists:
for repo in event_list:
name = repo.repo.name
for event in repo.contribs.nodes:
repo_contribs[name].add(event_type, event)
if repo.contribs.page_info.has_next_page:
after = repo.contribs.page_info.end_cursor
cont = True
for repo in response.repo_contribs.nodes:
name = repo.repo.name
node = RawEventsEvent(occurredAt=repo.occurred_at)
repo_contribs[name].add("repo", node)
return repo_contribs
def get_all_commit_info(
user_id: str,
name_with_owner: str,
start_date: datetime,
end_date: datetime,
access_token: Optional[str] = None,
) -> List[RESTRawCommit]:
owner, repo = name_with_owner.split("/")
data: List[RESTRawCommit] = []
for i in range(10):
if len(data) == 100 * i:
new_data = get_repo_commits(
owner, repo, user_id, start_date, end_date, i + 1, access_token
)
data.extend(new_data)
# sort ascending
return sorted(data, key=lambda x: x.timestamp)
async def get_all_commit_languages(
commit_infos: List[List[RESTRawCommit]],
repos: List[str],
repo_infos: Dict[str, RawRepo],
access_token: Optional[str] = None,
catch_errors: bool = False,
) -> Tuple[Dict[str, List[datetime]], Dict[str, List[CommitLanguages]]]:
commit_node_ids = [[x.node_id for x in repo] for repo in commit_infos]
commit_times = [[x.timestamp for x in repo] for repo in commit_infos]
id_mapping: Dict[str, Tuple[int, int]] = {}
repo_mapping: Dict[str, str] = {}
all_node_ids: List[str] = []
for i, repo_node_ids in enumerate(commit_node_ids):
for j, node_id in enumerate(repo_node_ids):
id_mapping[node_id] = (i, j)
repo_mapping[node_id] = repos[i]
all_node_ids.append(node_id)
node_id_chunks: List[List[str]] = [
all_node_ids[i : min(len(all_node_ids), i + GRAPHQL_NODE_CHUNK_SIZE)]
for i in range(0, len(all_node_ids), GRAPHQL_NODE_CHUNK_SIZE)
]
commit_language_chunks: List[List[Optional[GraphQLRawCommit]]] = await gather(
funcs=[get_commits for _ in node_id_chunks],
args_dicts=[
{
"node_ids": node_id_chunk,
"access_token": access_token,
"catch_errors": catch_errors,
}
for node_id_chunk in node_id_chunks
],
max_threads=GRAPHQL_NODE_THREADS,
)
temp_commit_languages: List[Optional[GraphQLRawCommit]] = []
for commit_language_chunk in commit_language_chunks:
temp_commit_languages.extend(commit_language_chunk)
# returns commits with no associated PR or incomplete PR
filtered_commits: List[GraphQLRawCommit] = filter(
lambda x: x is not None
and (len(x.prs.nodes) == 0 or x.prs.nodes[0].changed_files > PR_FILES)
and (x.additions + x.deletions > 100),
temp_commit_languages,
) # type: ignore
# get NODE_QUERIES largest commits with no associated PR or incomplete PR
sorted_commits = sorted(
filtered_commits, key=lambda x: x.additions + x.deletions, reverse=True
)[:NODE_QUERIES]
sorted_commit_urls = [commit.url.split("/") for commit in sorted_commits]
commit_files: List[List[RawCommitFile]] = await gather(
funcs=[get_commit_files for _ in sorted_commit_urls],
args_dicts=[
{
"owner": url[3],
"repo": url[4],
"sha": url[6],
"access_token": access_token,
}
for url in sorted_commit_urls
],
max_threads=REST_NODE_THREADS,
)
commit_files_dict: Dict[str, List[RawCommitFile]] = {
commit.url: commit_file
for commit, commit_file in zip(sorted_commits, commit_files)
}
commit_languages: List[List[CommitLanguages]] = [
[CommitLanguages() for _ in repo] for repo in commit_infos
]
for raw_commits, node_ids in zip(commit_language_chunks, node_id_chunks):
for raw_commit, node_id in zip(raw_commits, node_ids):
curr_commit_files: Optional[List[RawCommitFile]] = None
if raw_commit is not None and raw_commit.url in commit_files_dict:
curr_commit_files = commit_files_dict[raw_commit.url]
lang_breakdown = get_commit_languages(
raw_commit, curr_commit_files, repo_infos[repo_mapping[node_id]]
)
i, j = id_mapping[node_id]
commit_languages[i][j] = lang_breakdown
commit_times_dict: Dict[str, List[datetime]] = {}
commit_languages_dict: Dict[str, List[CommitLanguages]] = {}
for repo, times, languages in zip(repos, commit_times, commit_languages):
commit_times_dict[repo] = times
commit_languages_dict[repo] = languages
return commit_times_dict, commit_languages_dict
async def get_cleaned_contributions(
user_id: str,
start_date: datetime,
end_date: datetime,
access_token: Optional[str],
catch_errors: bool = False,
) -> Tuple[
RawCalendar,
Dict[str, ContribsList],
Dict[str, RawRepo],
Dict[str, List[datetime]],
Dict[str, List[CommitLanguages]],
]:
calendar = get_user_contribution_calendar(
user_id, start_date, end_date, access_token
)
contrib_events = get_user_all_contribution_events(
user_id, start_date, end_date, access_token
)
repos: List[str] = list(set(contrib_events.keys()))
commit_infos: List[List[RESTRawCommit]] = await gather(
funcs=[get_all_commit_info for _ in repos],
args_dicts=[
{
"user_id": user_id,
"name_with_owner": repo,
"start_date": start_date,
"end_date": end_date,
"access_token": access_token,
}
for repo in repos
],
max_threads=REST_NODE_THREADS,
)
_repo_infos: List[Optional[RawRepo]] = await gather(
funcs=[get_repo for _ in repos],
args_dicts=[
{
"owner": repo.split("/")[0],
"repo": repo.split("/")[1],
"access_token": access_token,
"catch_errors": catch_errors,
}
for repo in repos
],
max_threads=REST_NODE_THREADS,
)
repo_infos = {k: v for k, v in zip(repos, _repo_infos) if v is not None}
commit_times_dict, commit_languages_dict = await get_all_commit_languages(
commit_infos,
repos,
repo_infos,
access_token,
catch_errors,
)
return (
calendar,
contrib_events,
repo_infos,
commit_times_dict,
commit_languages_dict,
)
class StatsContainer:
def __init__(self):
self.contribs: int = 0
self.commits: int = 0
self.issues: int = 0
self.prs: int = 0
self.reviews: int = 0
self.repos: int = 0
self.other: int = 0
self.languages = CommitLanguages()
def add_stat(self, label: str, count: int, add: bool = False) -> None:
if label == "commit":
self.commits += count
elif label == "issue":
self.issues += count
elif label == "pr":
self.prs += count
elif label == "review":
self.reviews += count
elif label == "repo":
self.repos += count
if add:
self.contribs += count
else:
self.other -= count
def to_dict(self) -> Dict[str, Any]:
return {
"contribs_count": self.contribs,
"commits_count": self.commits,
"issues_count": self.issues,
"prs_count": self.prs,
"reviews_count": self.reviews,
"repos_count": self.repos,
"other_count": self.other,
"languages": self.languages.to_dict(),
}
class ListsContainer:
def __init__(self):
self.commits: List[datetime] = []
self.issues: List[datetime] = []
self.prs: List[datetime] = []
self.reviews: List[datetime] = []
self.repos: List[datetime] = []
def add_list(self, label: str, times: List[datetime]) -> None:
if label == "commit":
self.commits.extend(times)
elif label == "issue":
self.issues.extend(times)
elif label == "pr":
self.prs.extend(times)
elif label == "review":
self.reviews.extend(times)
elif label == "repo":
self.repos.extend(times)
def to_dict(self) -> Dict[str, Any]:
return {
"commits": self.commits,
"issues": self.issues,
"prs": self.prs,
"reviews": self.reviews,
"repos": self.repos,
}
class DateContainer:
def __init__(self):
self.date = ""
self.weekday = 0
self.stats = StatsContainer()
self.lists = ListsContainer()
def add_stat(
self, label: str, count: int, times: List[datetime], add: bool = False
):
self.stats.add_stat(label, count, add)
self.lists.add_list(label, times)
def to_dict(self) -> Dict[str, Any]:
return {
"date": self.date,
"weekday": self.weekday,
"stats": self.stats.to_dict(),
"lists": self.lists.to_dict(),
}
# assumed one month span, can be no more than one year
async def get_contributions(
user_id: str,
start_date: date,
end_date: date,
timezone_str: str = "US/Eastern",
access_token: Optional[str] = None,
catch_errors: bool = False,
) -> UserContributions:
tz = pytz.timezone(timezone_str)
start_month = date_to_datetime(start_date)
end_month = date_to_datetime(end_date, hour=23, minute=59, second=59)
(
calendar,
contrib_events,
repo_infos,
commit_times_dict,
commit_languages_dict,
) = await get_cleaned_contributions(
user_id, start_month, end_month, access_token, catch_errors
)
total_stats = StatsContainer()
public_stats = StatsContainer()
total: Dict[str, DateContainer] = defaultdict(DateContainer)
public: Dict[str, DateContainer] = defaultdict(DateContainer)
repo_stats: Dict[str, StatsContainer] = defaultdict(StatsContainer)
repositories: Dict[str, Dict[str, DateContainer]] = defaultdict(
lambda: defaultdict(DateContainer)
)
for week in calendar.weeks:
for day in week.contribution_days:
day_str = str(day.date)
for obj, stats_obj in [(total, total_stats), (public, public_stats)]:
obj[day_str].date = day.date.isoformat()
obj[day_str].weekday = day.weekday
obj[day_str].stats.contribs = day.count
obj[day_str].stats.other = day.count
stats_obj.contribs += day.count
stats_obj.other += day.count
def update_stats(
date_str: str, repo: str, event: str, count: int, times: List[datetime]
):
# update global counts for this event
total[date_str].add_stat(event, count, times)
total_stats.add_stat(event, count)
if not repo_infos[repo].is_private:
public[date_str].add_stat(event, count, times)
public_stats.add_stat(event, count)
repositories[repo][date_str].add_stat(event, count, times, add=True)
repo_stats[repo].add_stat(event, count, add=True)
def update_langs(date_str: str, repo: str, langs: CommitLanguages):
stores = [
total[date_str].stats.languages,
total_stats.languages,
repositories[repo][date_str].stats.languages,
repo_stats[repo].languages,
]
if not repo_infos[repo].is_private:
stores.append(public[date_str].stats.languages)
stores.append(public_stats.languages)
for store in stores:
store += langs
for repo, repo_events in contrib_events.items():
for label, events in [
("commit", repo_events.commits),
("issue", repo_events.issues),
("pr", repo_events.prs),
("review", repo_events.reviews),
("repo", repo_events.repos),
]:
events = sorted(events, key=lambda x: x.occurred_at)
for event in events:
datetime_obj = event.occurred_at.astimezone(tz)
date_str = datetime_obj.date().isoformat()
repositories[repo][date_str].date = date_str
if isinstance(event, RawEventsCommit):
count = 0
commit_times: List[datetime] = []
while len(commit_languages_dict[repo]) > 0 and count < event.count:
commit_times.append(commit_times_dict[repo].pop(0))
langs = commit_languages_dict[repo].pop(0)
update_langs(date_str, repo, langs)
count += 1
update_stats(date_str, repo, "commit", event.count, commit_times)
else:
update_stats(date_str, repo, label, 1, [datetime_obj])
total_stats_dict = total_stats.to_dict()
public_stats_dict = public_stats.to_dict()
repo_stats_dict = {name: stats.to_dict() for name, stats in repo_stats.items()}
for repo in repo_stats:
repo_stats_dict[repo]["private"] = repo_infos[repo].is_private
total_list = [v.to_dict() for v in total.values() if v.stats.contribs > 0]
public_list = [v.to_dict() for v in public.values() if v.stats.contribs > 0]
repositories_list = {
name: [v.to_dict() for v in repo.values()]
for name, repo in repositories.items()
}
output = UserContributions.model_validate(
{
"total_stats": total_stats_dict,
"public_stats": public_stats_dict,
"total": total_list,
"public": public_list,
"repo_stats": repo_stats_dict,
"repos": repositories_list,
}
)
return output
@@ -1,39 +0,0 @@
from typing import List, Optional
from src.data.github.graphql import (
get_user_followers as _get_user_followers,
get_user_following as _get_user_following,
)
from src.models import User, UserFollows
def get_user_follows(user_id: str, access_token: Optional[str]) -> UserFollows:
"""get user followers and users following for given user"""
followers: List[User] = []
following: List[User] = []
for user_list, get_func in zip(
[followers, following], [_get_user_followers, _get_user_following]
):
after: Optional[str] = ""
index, cont = 0, True # initialize variables
while cont and index < 10:
after_str: str = after if isinstance(after, str) else ""
data = get_func(user_id, after=after_str, access_token=access_token)
cont = False
user_list.extend(
map(
lambda x: User(name=x.name, login=x.login, url=x.url),
data.nodes,
)
)
if data.page_info.has_next_page:
after = data.page_info.end_cursor
cont = True
index += 1
return UserFollows(followers=followers, following=following)
@@ -1,118 +0,0 @@
from json import load
from typing import Any, Dict, List, Optional, Union
from src.constants import BLACKLIST, CUTOFF, DEFAULT_COLOR, FILE_CUTOFF
from src.data.github.graphql import RawCommit, RawRepo
from src.data.github.rest import RawCommitFile
EXTENSIONS: Dict[str, Dict[str, str]] = load(open("./src/data/github/extensions.json"))
class CommitLanguages:
def __init__(self):
self.langs: Dict[str, Dict[str, Union[str, int]]] = {}
def __repr__(self):
return str(self.langs)
def add_lines(
self, name: str, color: Optional[str], additions: int, deletions: int
):
if (
name not in BLACKLIST
and max(additions, deletions) > 0
and max(additions, deletions) < FILE_CUTOFF
):
color = color or DEFAULT_COLOR
if name not in self.langs:
self.langs[name] = {"color": color, "additions": 0, "deletions": 0}
self.langs[name]["additions"] += additions # type: ignore
self.langs[name]["deletions"] += deletions # type: ignore
def normalize(self, add_ratio: float, del_ratio: float):
for lang in self.langs:
new_add = round(int(self.langs[lang]["additions"]) * add_ratio)
self.langs[lang]["additions"] = new_add
new_del = round(int(self.langs[lang]["deletions"]) * del_ratio)
self.langs[lang]["deletions"] = new_del
def __add__(self, other: "CommitLanguages"):
for lang in other.langs:
if lang not in self.langs:
self.langs[lang] = other.langs[lang].copy()
else:
self.langs[lang]["additions"] += other.langs[lang]["additions"] # type: ignore
self.langs[lang]["deletions"] += other.langs[lang]["deletions"] # type: ignore
def to_dict(self) -> Dict[str, Any]:
return self.langs
def get_commit_languages(
commit: Optional[RawCommit],
files: Optional[List[RawCommitFile]],
repo: RawRepo,
) -> CommitLanguages:
out = CommitLanguages()
if commit is None:
return out
if max(commit.additions, commit.deletions) == 0:
return out
# assummed to be auto-generated or copied
if max(commit.additions, commit.deletions) > 10 * CUTOFF or (
max(commit.additions, commit.deletions) > CUTOFF
and min(commit.additions, commit.deletions) == 0
):
return out
pr_coverage = 0
if len(commit.prs.nodes) > 0:
pr_obj = commit.prs.nodes[0]
pr_files = pr_obj.files.nodes
total_changed = sum(x.additions + x.deletions for x in pr_files)
pr_coverage = total_changed / max(1, (pr_obj.additions + pr_obj.deletions))
if files is not None:
for file in files:
filename = file.filename.split(".")
extension = "" if len(filename) <= 1 else filename[-1]
lang = EXTENSIONS.get(f".{extension}", None)
if lang is not None:
out.add_lines(
lang["name"], lang["color"], file.additions, file.deletions
)
elif len(commit.prs.nodes) > 0 and pr_coverage > 0.25:
pr = commit.prs.nodes[0]
total_additions, total_deletions = 0, 0
for file in pr.files.nodes:
filename = file.path.split(".")
extension = "" if len(filename) <= 1 else filename[-1]
lang = EXTENSIONS.get(f".{extension}", None)
if lang is not None:
out.add_lines(
lang["name"], lang["color"], file.additions, file.deletions
)
total_additions += file.additions
total_deletions += file.deletions
add_ratio = min(pr.additions, commit.additions) / max(1, total_additions)
del_ratio = min(pr.deletions, commit.deletions) / max(1, total_deletions)
out.normalize(add_ratio, del_ratio)
elif commit.additions + commit.deletions > 2 * CUTOFF:
# assummed to be auto generated
return out
else:
repo_info = repo.languages.edges
languages = [x for x in repo_info if x.node.name not in BLACKLIST]
total_repo_size = sum(language.size for language in languages)
for language in languages:
lang_name = language.node.name
lang_color = language.node.color
lang_size = language.size
additions = round(commit.additions * lang_size / total_repo_size)
deletions = round(commit.deletions * lang_size / total_repo_size)
out.add_lines(lang_name, lang_color, additions, deletions)
return out
@@ -1,28 +0,0 @@
from datetime import date
from typing import Optional
from src.aggregation.layer0.contributions import get_contributions
from src.models import UserPackage
# from src.subscriber.aggregation.user.follows import get_user_follows
async def get_user_data(
user_id: str,
start_date: date,
end_date: date,
timezone_str: str,
access_token: Optional[str],
catch_errors: bool = False,
) -> UserPackage:
"""packages all processing steps for the user query"""
contribs = await get_contributions(
user_id=user_id,
start_date=start_date,
end_date=end_date,
timezone_str=timezone_str,
access_token=access_token,
catch_errors=catch_errors,
)
return UserPackage(contribs=contribs)
@@ -1,3 +0,0 @@
from src.aggregation.layer1.user import query_user
__all__ = ["query_user"]
@@ -1,61 +0,0 @@
from datetime import timedelta
from typing import List, Optional, Tuple
from src.constants import OWNER, REPO
from src.data.github.rest import (
RESTError,
RESTErrorNotFound,
get_repo_stargazers as github_get_repo_stargazers,
get_user as github_get_user,
get_user_starred_repos as github_get_user_starred_repos,
)
from src.data.github.utils import get_access_token
from src.data.mongo.user import get_public_user as db_get_public_user
from src.utils import alru_cache
async def get_valid_github_user(user_id: str) -> Optional[str]:
access_token = get_access_token()
try:
return github_get_user(user_id, access_token)["login"]
except (RESTErrorNotFound, KeyError):
# User does not exist
return None
except RESTError:
# Rate limited, so assume user exists
return user_id
async def get_valid_db_user(user_id: str) -> bool:
user = await db_get_public_user(user_id)
return user is not None
@alru_cache(ttl=timedelta(minutes=15))
async def get_repo_stargazers(
owner: str = OWNER, repo: str = REPO, no_cache: bool = False
) -> Tuple[bool, List[str]]:
access_token = get_access_token()
data: List[str] = []
page = 0
while len(data) == 100 * page:
temp_data = github_get_repo_stargazers(access_token, owner, repo, page=page)
temp_data = [x["user"]["login"] for x in temp_data]
data.extend(temp_data)
page += 1
return (True, data)
async def get_user_stars(user_id: str) -> List[str]:
access_token = get_access_token()
try:
data = github_get_user_starred_repos(user_id, access_token)
data = [x["repo"]["full_name"] for x in data]
return data
except RESTErrorNotFound:
# User does not exist (and rate limited previously)
return []
except RESTError:
# Rate limited, so assume user starred repo
return [f"{OWNER}/{REPO}"]
@@ -1,125 +0,0 @@
from calendar import monthrange
from datetime import date, datetime, timedelta
from typing import List, Optional, Tuple
import requests
from src.aggregation.layer0.package import get_user_data
from src.constants import API_VERSION # , BACKEND_URL, PROD
from src.data.github.graphql import GraphQLErrorRateLimit
from src.data.mongo.secret import update_keys
from src.data.mongo.user_months import UserMonth, get_user_months, set_user_month
from src.models.user.main import UserPackage
from src.utils import alru_cache, date_to_datetime
s = requests.Session()
# Formerly the subscriber, compute and save new data here
async def query_user_month(
user_id: str,
access_token: Optional[str],
private_access: bool,
start_date: date,
retries: int = 0,
) -> Optional[UserMonth]:
year, month = start_date.year, start_date.month
end_day = monthrange(year, month)[1]
end_date = date(year, month, end_day)
try:
data = await get_user_data(
user_id=user_id,
start_date=start_date,
end_date=end_date,
timezone_str="US/Eastern",
access_token=access_token,
catch_errors=retries > 0,
)
except GraphQLErrorRateLimit:
return None
except Exception:
# Retry, catching exceptions and marking incomplete this time
if retries < 1:
await query_user_month(
user_id, access_token, private_access, start_date, retries + 1
)
return None
month_completed = datetime.now() > date_to_datetime(end_date) + timedelta(days=1)
user_month = UserMonth.model_validate(
{
"user_id": user_id,
"month": date_to_datetime(start_date),
"version": API_VERSION,
"private": private_access,
"complete": retries == 0 and month_completed,
"data": data,
}
)
await set_user_month(user_month)
return user_month
@alru_cache(ttl=timedelta(hours=6))
async def query_user(
user_id: str,
access_token: Optional[str],
private_access: bool = False,
start_date: date = date.today() - timedelta(365),
end_date: date = date.today(),
max_time: int = 3600, # seconds
no_cache: bool = False,
) -> Tuple[bool, UserPackage]:
# Return (possibly incomplete) within max_time seconds
start_time = datetime.now()
incomplete = False
await update_keys()
curr_data: List[UserMonth] = await get_user_months(
user_id, private_access, start_date, end_date
)
curr_months = [x.month for x in curr_data if x.complete]
month, year = start_date.month, start_date.year
new_months: List[date] = []
while date(year, month, 1) <= end_date:
start = date(year, month, 1)
if date_to_datetime(start) not in curr_months:
new_months.append(start)
month = month % 12 + 1
year = year + (month == 1)
# Start with complete months and add any incomplete months
all_user_packages: List[UserPackage] = [x.data for x in curr_data if x.complete]
for month in new_months:
if datetime.now() - start_time < timedelta(seconds=max_time):
temp = await query_user_month(user_id, access_token, private_access, month)
if temp is not None:
all_user_packages.append(temp.data)
else:
incomplete = True
else:
incomplete = True
out: UserPackage = UserPackage.empty()
if len(all_user_packages) > 0:
out = all_user_packages[0]
for user_package in all_user_packages[1:]:
out += user_package
out.incomplete = incomplete
if incomplete or len(new_months) > 1:
# TODO: figure out why this causes an infinite loop
# # cache buster for publisher
# if PROD:
# s.get(f"{BACKEND_URL}/user/{user_id}?no_cache=True")
return (False, out)
# only cache if just the current month updated
return (True, out)
@@ -1,4 +0,0 @@
from src.aggregation.layer2.auth import get_is_valid_user
from src.aggregation.layer2.user import get_user, get_user_demo
__all__ = ["get_is_valid_user", "get_user", "get_user_demo"]
@@ -1,58 +0,0 @@
from datetime import timedelta
from typing import Optional, Tuple
from src.aggregation.layer1.auth import (
get_repo_stargazers,
get_user_stars,
get_valid_db_user,
get_valid_github_user,
)
from src.constants import OWNER, REPO, USER_BLACKLIST, USER_WHITELIST
from src.data.github.rest import RESTError
from src.utils import alru_cache
async def check_github_user_exists(user_id: str) -> Optional[str]:
return await get_valid_github_user(user_id)
async def check_db_user_exists(user_id: str) -> bool:
return await get_valid_db_user(user_id)
async def check_user_starred_repo(
user_id: str, owner: str = OWNER, repo: str = REPO
) -> bool:
return True
# Checks the repo's starred users (with cache)
try:
repo_stargazers = await get_repo_stargazers(owner, repo)
if user_id in repo_stargazers:
return True
except RESTError:
return True # Assume the user has starred the repo
# Checks the user's 30 most recent starred repos (no cache)
user_stars = await get_user_stars(user_id)
return f"{owner}/{repo}" in user_stars
@alru_cache(ttl=timedelta(hours=1))
async def get_is_valid_user(user_id: str) -> Tuple[bool, str]:
if user_id.lower() in USER_BLACKLIST:
# TODO: change error message
return (False, "GitHub user not found")
if user_id.lower() in USER_WHITELIST:
return (True, f"Valid user {user_id.lower()}")
valid_user_id = await check_github_user_exists(user_id)
if valid_user_id is None:
return (False, "GitHub user not found")
valid_db_user = await check_db_user_exists(valid_user_id)
user_starred = await check_user_starred_repo(valid_user_id)
if not (user_starred or valid_db_user):
return (False, "Repo not starred")
return (True, f"Valid user {valid_user_id}")
@@ -1,79 +0,0 @@
from datetime import date, timedelta
from typing import Optional, Tuple
from src.aggregation.layer0 import get_user_data
from src.constants import USER_BLACKLIST
from src.data.mongo.secret.functions import update_keys
from src.data.mongo.user import PublicUserModel, get_public_user as db_get_public_user
from src.data.mongo.user_months import get_user_months
from src.models import UserPackage
from src.models.background import UpdateUserBackgroundTask
from src.utils import alru_cache
# Formerly the publisher, loads existing data here
async def _get_user(
user_id: str, private_access: bool, start_date: date, end_date: date
) -> Tuple[Optional[UserPackage], bool]:
user_months = await get_user_months(user_id, private_access, start_date, end_date)
if len(user_months) == 0:
return None, False
expected_num_months = (
(end_date.year - start_date.year) * 12 + (end_date.month - start_date.month) + 1
)
complete = len(user_months) == expected_num_months
user_data = user_months[0].data
for user_month in user_months[1:]:
user_data += user_month.data
# TODO: handle timezone_str here
return user_data.trim(start_date, end_date), complete
@alru_cache()
async def get_user(
user_id: str,
start_date: date,
end_date: date,
no_cache: bool = False,
) -> Tuple[
bool, Tuple[Optional[UserPackage], bool, Optional[UpdateUserBackgroundTask]]
]:
if user_id in USER_BLACKLIST:
return (False, (None, False, None))
user: Optional[PublicUserModel] = await db_get_public_user(user_id)
if user is None:
return (False, (None, False, None))
private_access = user.private_access or False
user_data, complete = await _get_user(user_id, private_access, start_date, end_date)
background_task = UpdateUserBackgroundTask(
user_id=user_id,
access_token=user.access_token,
private_access=private_access,
start_date=start_date,
end_date=end_date,
)
return (complete, (user_data, complete, background_task))
@alru_cache(ttl=timedelta(minutes=15))
async def get_user_demo(
user_id: str, start_date: date, end_date: date, no_cache: bool = False
) -> Tuple[bool, UserPackage]:
await update_keys()
timezone_str = "US/Eastern"
# recompute/cache but don't save to db
data = await get_user_data(
user_id=user_id,
start_date=start_date,
end_date=end_date,
timezone_str=timezone_str,
access_token=None,
catch_errors=True,
)
return (True, data)
-74
View File
@@ -1,74 +0,0 @@
import os
# GLOBAL
LOCAL = os.getenv("LOCAL", "False") == "True"
PROD = os.getenv("PROD", "False") == "True"
PROJECT_ID = "github-334619"
BACKEND_URL = "https://api.githubtrends.io" if PROD else "http://localhost:8000"
OWNER = "avgupta456"
REPO = "github-trends"
# API
# https://docs.github.com/en/rest/reference/rate-limit
# https://docs.github.com/en/rest/guides/best-practices-for-integrators#dealing-with-secondary-rate-limits
# https://docs.github.com/en/graphql/overview/resource-limitations
TIMEOUT = 15 # max seconds to wait for api response
GRAPHQL_NODE_CHUNK_SIZE = 50 # number of nodes (commits) to query (max 100)
GRAPHQL_NODE_THREADS = 5 # number of node queries simultaneously (avoid blacklisting)
REST_NODE_THREADS = 50 # number of node queries simultaneously (avoid blacklisting)
PR_FILES = 5 # max number of files to query for PRs
NODE_QUERIES = 20 # max number of node queries to make
CUTOFF = 1000 # if additions or deletions > CUTOFF, or sum > 2 * CUTOFF, ignore LOC
FILE_CUTOFF = 1000 # if less than cutoff in file, count LOC
API_VERSION = 0.02 # determines when to overwrite MongoDB data
# CUSTOMIZATION
BLACKLIST = ["Jupyter Notebook", "HTML"] # languages to ignore
# OAUTH
prefix = "PROD" if PROD else "DEV"
# client ID for GitHub OAuth App
OAUTH_CLIENT_ID = os.getenv(f"{prefix}_OAUTH_CLIENT_ID", "")
# client secret for App
OAUTH_CLIENT_SECRET = os.getenv(f"{prefix}_OAUTH_CLIENT_SECRET", "")
# redirect uri for App
OAUTH_REDIRECT_URI = os.getenv(f"{prefix}_OAUTH_REDIRECT_URI", "")
# MONGODB
MONGODB_PASSWORD = os.getenv("MONGODB_PASSWORD", "")
# SVG
DEFAULT_COLOR = "#858585"
# SENTRY
SENTRY_DSN = os.getenv("SENTRY_DSN", "")
# TESTING
TEST_USER_ID = "avgupta456"
TEST_REPO = "github-trends"
TEST_TOKEN = os.getenv("AUTH_TOKEN", "") # for authentication
TEST_NODE_IDS = [
"C_kwDOENp939oAKGM1MzdlM2QzMTZjMmEyZGIyYWU4ZWI0MmNmNjQ4YWEwNWQ5OTBiMjM",
"C_kwDOD_-BVNoAKDFhNTIxNWE1MGM4ZDllOGEwYTFhNjhmYWZkYzE5MzA5YTRkMDMwZmM",
"C_kwDOD_-BVNoAKDRiZTQ4MTQ0MzgwYjBlNGEwNjQ4YjY4YWI4ZjFjYmQ3MWU4M2VhMzU",
]
TEST_SHA = "ad83e6340377904fa0295745b5314202b23d2f3f"
# WRAPPED
# example users, don't need to star the repo
USER_WHITELIST = [
"torvalds",
"yyx990803",
"shadcn",
"sindresorhus",
]
USER_BLACKLIST = ["kangmingtay", "ae7er", "stalukdar7", "piyush7833"]
print("PROD", PROD)
print("API_VERSION", API_VERSION)
print()
@@ -1,3 +0,0 @@
from src.data.github.auth.main import authenticate
__all__ = ["authenticate"]
@@ -1,57 +0,0 @@
from datetime import datetime
from typing import Dict, Optional, Tuple
import requests
from src.constants import OAUTH_CLIENT_ID, OAUTH_CLIENT_SECRET, OAUTH_REDIRECT_URI
s = requests.session()
def get_unknown_user(access_token: str) -> Optional[str]:
"""
Accepts access_token and returns user_id of associated user
:param access_token: GitHub access token
:return: user_id or None if invalid access_token
"""
headers: Dict[str, str] = {
"Accept": "application/vnd.github.v3+json",
"Authorization": f"bearer {access_token}",
}
r = s.get("https://api.github.com/user", params={}, headers=headers)
return r.json().get("login", None)
class OAuthError(Exception):
pass
async def authenticate(code: str) -> Tuple[str, str]:
"""
Takes a authentication code, verifies, and returns user_id/access_token
:param code: GitHub authentication code from OAuth process
:return: user_id, access_token of authenticated user
"""
start = datetime.now()
params = {
"client_id": OAUTH_CLIENT_ID,
"client_secret": OAUTH_CLIENT_SECRET,
"code": code,
"redirect_uri": OAUTH_REDIRECT_URI,
}
r = s.post("https://github.com/login/oauth/access_token", params=params)
if r.status_code != 200:
raise OAuthError(f"OAuth Error: {str(r.status_code)}")
access_token = r.text.split("&")[0].split("=")[1]
user_id = get_unknown_user(access_token)
if user_id is None:
raise OAuthError("OAuth Error: Invalid user_id/access_token")
print("OAuth SignUp", datetime.now() - start)
return user_id, access_token
File diff suppressed because it is too large Load Diff
@@ -1,44 +0,0 @@
from src.data.github.graphql.commit import get_commits
from src.data.github.graphql.models import RawCommit, RawRepo
from src.data.github.graphql.repo import get_repo
from src.data.github.graphql.template import (
GraphQLErrorMissingNode,
GraphQLErrorRateLimit,
GraphQLErrorTimeout,
get_query_limit,
)
from src.data.github.graphql.user.contribs.contribs import (
get_user_contribution_calendar,
get_user_contribution_events,
)
from src.data.github.graphql.user.contribs.models import (
RawCalendar,
RawEvents,
RawEventsCommit,
RawEventsEvent,
)
from src.data.github.graphql.user.follows.follows import (
get_user_followers,
get_user_following,
)
from src.data.github.graphql.user.follows.models import RawFollows
__all__ = [
"get_commits",
"RawCommit",
"RawRepo",
"get_repo",
"GraphQLErrorMissingNode",
"GraphQLErrorRateLimit",
"GraphQLErrorTimeout",
"get_query_limit",
"get_user_contribution_calendar",
"get_user_contribution_events",
"RawCalendar",
"RawEvents",
"RawEventsCommit",
"RawEventsEvent",
"get_user_followers",
"get_user_following",
"RawFollows",
]
@@ -1,95 +0,0 @@
from typing import List, Optional
from src.constants import PR_FILES
from src.data.github.graphql.models import RawCommit
from src.data.github.graphql.template import (
GraphQLError,
GraphQLErrorMissingNode,
GraphQLErrorRateLimit,
GraphQLErrorTimeout,
get_template,
)
def get_commits(
node_ids: List[str], access_token: Optional[str] = None, catch_errors: bool = False
) -> List[Optional[RawCommit]]:
"""
Gets all repository data from graphql
:param access_token: GitHub access token
:param node_ids: List of node ids
:return: List of commits
"""
if PR_FILES == 0: # type: ignore
query = {
"variables": {"ids": node_ids},
"query": """
query getCommits($ids: [ID!]!) {
nodes(ids: $ids) {
... on Commit {
additions
deletions
changedFiles
url
}
}
}
""",
}
else:
query = {
"variables": {"ids": node_ids, "first": PR_FILES},
"query": """
query getCommits($ids: [ID!]!, $first: Int!) {
nodes(ids: $ids) {
... on Commit {
additions
deletions
changedFiles
url
associatedPullRequests(first: 1) {
nodes {
changedFiles
additions
deletions
files(first: $first) {
nodes {
path
additions
deletions
}
}
}
}
}
}
}
""",
}
try:
raw_commits = get_template(query, access_token)["data"]["nodes"]
except GraphQLErrorMissingNode as e:
return (
get_commits(node_ids[: e.node], access_token)
+ [None]
+ get_commits(node_ids[e.node + 1 :], access_token)
)
except (GraphQLErrorRateLimit, GraphQLErrorTimeout, GraphQLError) as e:
if catch_errors:
return [None for _ in node_ids]
raise e
out: List[Optional[RawCommit]] = []
for raw_commit in raw_commits:
try:
if "associatedPullRequests" not in raw_commit:
raw_commit["associatedPullRequests"] = {"nodes": []}
out.append(RawCommit.model_validate(raw_commit))
except Exception as e:
if catch_errors:
out.append(None)
else:
raise e
return out
@@ -1,55 +0,0 @@
from typing import List, Optional
from pydantic import BaseModel, Field
class RawCommitPRFileNode(BaseModel):
path: str
additions: int
deletions: int
class RawCommitPRFile(BaseModel):
nodes: List[RawCommitPRFileNode]
class RawCommitPRNode(BaseModel):
changed_files: int = Field(alias="changedFiles")
additions: int
deletions: int
files: RawCommitPRFile
class RawCommitPR(BaseModel):
nodes: List[RawCommitPRNode]
class RawCommit(BaseModel):
additions: int
deletions: int
changed_files: int = Field(alias="changedFiles")
url: str
prs: RawCommitPR = Field(alias="associatedPullRequests")
class RawRepoLanguageNode(BaseModel):
name: str
color: Optional[str]
class RawRepoLanguageEdge(BaseModel):
node: RawRepoLanguageNode
size: int
class RawRepoLanguage(BaseModel):
total_count: int = Field(alias="totalCount")
total_size: int = Field(alias="totalSize")
edges: List[RawRepoLanguageEdge]
class RawRepo(BaseModel):
is_private: bool = Field(alias="isPrivate")
fork_count: int = Field(alias="forkCount")
stargazer_count: int = Field(alias="stargazerCount")
languages: RawRepoLanguage
@@ -1,50 +0,0 @@
from typing import Optional
from src.data.github.graphql.models import RawRepo
from src.data.github.graphql.template import get_template
def get_repo(
owner: str,
repo: str,
access_token: Optional[str] = None,
catch_errors: bool = False,
) -> Optional[RawRepo]:
"""
Gets all repository data from graphql
:param access_token: GitHub access token
:param owner: GitHub owner
:param repo: GitHub repository
:return: RawRepo object or None if repo not present
"""
query = {
"variables": {"owner": owner, "repo": repo},
"query": """
query getRepo($owner: String!, $repo: String!) {
repository(owner: $owner, name: $repo) {
isPrivate,
forkCount,
stargazerCount,
languages(first: 10){
totalCount,
totalSize,
edges{
node {
name,
color,
},
size,
},
},
}
}
""",
}
try:
raw_repo = get_template(query, access_token)["data"]["repository"]
return RawRepo.model_validate(raw_repo)
except Exception as e:
if catch_errors:
return None
raise e
@@ -1,102 +0,0 @@
import logging
from datetime import datetime
from typing import Any, Dict, Optional, Tuple
import requests
from requests.exceptions import ReadTimeout
from src.constants import TIMEOUT
from src.data.github.utils import get_access_token
s = requests.session()
class GraphQLError(Exception):
pass
class GraphQLErrorMissingNode(Exception):
def __init__(self, node: int, *args: Tuple[Any], **kwargs: Dict[str, Any]):
super(Exception, self).__init__(*args, **kwargs)
self.node = node
class GraphQLErrorRateLimit(Exception):
pass
class GraphQLErrorTimeout(Exception):
pass
def get_template(
query: Dict[str, Any], access_token: Optional[str] = None, retries: int = 0
) -> Dict[str, Any]:
"""
Template for interacting with the GitHub GraphQL API
:param query: The query to be sent to the GitHub GraphQL API
:param access_token: The access token to be used for the query
:param retries: The number of retries to be made for Auth Exceptions
:return: The response from the GitHub GraphQL API
"""
start = datetime.now()
new_access_token = get_access_token(access_token)
headers: Dict[str, str] = {"Authorization": f"bearer {new_access_token}"}
try:
r = s.post(
"https://api.github.com/graphql",
json=query,
headers=headers,
timeout=TIMEOUT,
)
except ReadTimeout:
raise GraphQLErrorTimeout("GraphQL Error: Request Timeout")
print("GraphQL", new_access_token, datetime.now() - start)
if r.status_code == 200:
data = r.json()
if "errors" in data:
if (
"type" in data["errors"][0]
and data["errors"][0]["type"] in ["SERVICE_UNAVAILABLE", "NOT_FOUND"]
and "path" in data["errors"][0]
and isinstance(data["errors"][0]["path"], list)
and data["errors"][0]["path"][0] == "nodes"
):
raise GraphQLErrorMissingNode(node=int(data["errors"][0]["path"][1]))
if retries < 2:
print("GraphQL Error, Retrying:", new_access_token)
return get_template(query, access_token, retries + 1)
raise GraphQLError("GraphQL Error: " + str(data["errors"]))
return data
if r.status_code in [401, 403]:
if retries < 2:
print("GraphQL Error, Retrying:", new_access_token)
return get_template(query, access_token, retries + 1)
raise GraphQLErrorRateLimit("GraphQL Error: Unauthorized")
if r.status_code == 502:
raise GraphQLErrorTimeout("GraphQL Error: Request Timeout")
raise GraphQLError(f"GraphQL Error: {str(r.status_code)}")
def get_query_limit(access_token: str) -> int:
"""
Get the current rate limit for the GitHub GraphQL API
:param access_token: The access token to be used for the query
:return: The current rate limit for the GitHub GraphQL API
"""
try:
data = get_template(
{"query": "query { rateLimit { remaining } }"}, access_token
)
return data["data"]["rateLimit"]["remaining"]
except Exception as e:
logging.exception(e)
return -1
@@ -1,157 +0,0 @@
# import json
from datetime import datetime
from typing import Optional
from src.data.github.graphql.template import get_template
from src.data.github.graphql.user.contribs.models import RawCalendar, RawEvents
def get_user_contribution_calendar(
user_id: str,
start_date: datetime,
end_date: datetime,
access_token: Optional[str] = None,
) -> RawCalendar:
"""Gets contribution calendar for a given time period (max one year)"""
if (end_date - start_date).days > 365:
raise ValueError("date range can be at most 1 year")
query = {
"variables": {
"login": user_id,
"startDate": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
"endDate": end_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
},
"query": """
query getUser($login: String!, $startDate: DateTime!, $endDate: DateTime!){
user(login: $login){
contributionsCollection(from: $startDate, to: $endDate){
contributionCalendar{
weeks{
contributionDays{
date
weekday
contributionCount
}
}
}
}
}
}
""",
}
raw_data = get_template(query, access_token)
output = raw_data["data"]["user"]["contributionsCollection"]["contributionCalendar"]
return RawCalendar.model_validate(output)
def get_user_contribution_events(
user_id: str,
start_date: datetime,
end_date: datetime,
max_repos: int = 100,
first: int = 100,
after: str = "",
access_token: Optional[str] = None,
) -> RawEvents:
"""Fetches user contributions (commits, issues, prs, reviews)"""
query = {
"variables": {
"login": user_id,
"startDate": start_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
"endDate": end_date.strftime("%Y-%m-%dT%H:%M:%SZ"),
"maxRepos": max_repos,
"first": first,
"after": after,
},
"query": """
query getUser($login: String!, $startDate: DateTime!, $endDate: DateTime!, $maxRepos: Int!, $first: Int!, $after: String!) {
user(login: $login){
contributionsCollection(from: $startDate, to: $endDate){
commitContributionsByRepository(maxRepositories: $maxRepos){
repository{
nameWithOwner,
},
totalCount:contributions(first: 1){
totalCount
}
contributions(first: $first, after: $after){
nodes{
commitCount,
occurredAt,
}
pageInfo{
hasNextPage,
endCursor
}
}
}
issueContributionsByRepository(maxRepositories: $maxRepos){
repository{
nameWithOwner
},
totalCount:contributions(first: 1){
totalCount
}
contributions(first: $first, after: $after){
nodes{
occurredAt,
}
pageInfo{
hasNextPage,
endCursor
}
}
}
pullRequestContributionsByRepository(maxRepositories: $maxRepos){
repository{
nameWithOwner
},
totalCount:contributions(first: 1){
totalCount
}
contributions(first: $first, after: $after){
nodes{
occurredAt,
}
pageInfo{
hasNextPage,
endCursor
}
}
}
pullRequestReviewContributionsByRepository(maxRepositories: $maxRepos){
repository{
nameWithOwner
},
totalCount:contributions(first: 1){
totalCount
}
contributions(first: $first, after: $after){
nodes{
occurredAt,
}
pageInfo{
hasNextPage,
endCursor
}
}
},
repositoryContributions(first: $maxRepos){
totalCount
nodes{
repository{
nameWithOwner,
}
occurredAt,
}
},
},
}
}
""",
}
raw_data = get_template(query, access_token)
output = raw_data["data"]["user"]["contributionsCollection"]
return RawEvents.model_validate(output)
@@ -1,91 +0,0 @@
from datetime import date, datetime
from typing import List, Optional
from pydantic import BaseModel, Field
class RawCalendarDay(BaseModel):
date: date
weekday: int
count: int = Field(alias="contributionCount")
class RawCalendarWeek(BaseModel):
contribution_days: List[RawCalendarDay] = Field(alias="contributionDays")
class RawCalendar(BaseModel):
weeks: List[RawCalendarWeek]
class RawEventsRepoName(BaseModel):
name: str = Field(alias="nameWithOwner")
class RawEventsCount(BaseModel):
count: int = Field(alias="totalCount")
class RawEventsCommit(BaseModel):
count: int = Field(alias="commitCount")
occurred_at: datetime = Field(alias="occurredAt")
class RawEventsEvent(BaseModel):
occurred_at: datetime = Field(alias="occurredAt")
class RawEventsPageInfo(BaseModel):
has_next_page: bool = Field(alias="hasNextPage")
end_cursor: Optional[str] = Field(alias="endCursor")
class Config:
allow_none = True
class RawEventsCommits(BaseModel):
nodes: List[RawEventsCommit]
page_info: RawEventsPageInfo = Field(alias="pageInfo")
class RawEventsContribs(BaseModel):
nodes: List[RawEventsEvent]
page_info: RawEventsPageInfo = Field(alias="pageInfo")
class RawEventsRepoCommits(BaseModel):
repo: RawEventsRepoName = Field(alias="repository")
count: RawEventsCount = Field(alias="totalCount")
contribs: RawEventsCommits = Field(alias="contributions")
class RawEventsRepo(BaseModel):
repo: RawEventsRepoName = Field(alias="repository")
count: RawEventsCount = Field(alias="totalCount")
contribs: RawEventsContribs = Field(alias="contributions")
class RawEventsRepoEvent(BaseModel):
repo: RawEventsRepoName = Field(alias="repository")
occurred_at: datetime = Field(alias="occurredAt")
class RawEventsRepoContribs(BaseModel):
count: int = Field(alias="totalCount")
nodes: List[RawEventsRepoEvent]
class RawEvents(BaseModel):
commit_contribs_by_repo: List[RawEventsRepoCommits] = Field(
alias="commitContributionsByRepository"
)
issue_contribs_by_repo: List[RawEventsRepo] = Field(
alias="issueContributionsByRepository"
)
pr_contribs_by_repo: List[RawEventsRepo] = Field(
alias="pullRequestContributionsByRepository"
)
review_contribs_by_repo: List[RawEventsRepo] = Field(
alias="pullRequestReviewContributionsByRepository"
)
repo_contribs: RawEventsRepoContribs = Field(alias="repositoryContributions")
@@ -1,121 +0,0 @@
# import json
from typing import Dict, Optional, Union
from src.data.github.graphql.template import get_template
from src.data.github.graphql.user.follows.models import RawFollows
def get_user_followers(
user_id: str, first: int = 100, after: str = "", access_token: Optional[str] = None
) -> RawFollows:
"""gets user's followers and users following'"""
variables: Dict[str, Union[str, int]] = (
{"login": user_id, "first": first, "after": after}
if after != ""
else {"login": user_id, "first": first}
)
query_str: str = (
"""
query getUser($login: String!, $first: Int!, $after: String!) {
user(login: $login){
followers(first: $first, after: $after){
nodes{
name,
login,
url
}
pageInfo{
hasNextPage,
endCursor
}
}
}
}
"""
if after != ""
else """
query getUser($login: String!, $first: Int!) {
user(login: $login){
followers(first: $first){
nodes{
name,
login,
url
}
pageInfo{
hasNextPage,
endCursor
}
}
}
}
"""
)
query = {
"variables": variables,
"query": query_str,
}
output_dict = get_template(query, access_token)["data"]["user"]["followers"]
return RawFollows.model_validate(output_dict)
def get_user_following(
user_id: str, first: int = 10, after: str = "", access_token: Optional[str] = None
) -> RawFollows:
"""gets user's followers and users following'"""
variables: Dict[str, Union[str, int]] = (
{"login": user_id, "first": first, "after": after}
if after != ""
else {"login": user_id, "first": first}
)
query_str: str = (
"""
query getUser($login: String!, $first: Int!, $after: String!) {
user(login: $login){
following(first: $first, after: $after){
nodes{
name,
login,
url
}
pageInfo{
hasNextPage,
endCursor
}
}
}
}
"""
if after != ""
else """
query getUser($login: String!, $first: Int!) {
user(login: $login){
following(first: $first){
nodes{
name,
login,
url
}
pageInfo{
hasNextPage,
endCursor
}
}
}
}
"""
)
query = {
"variables": variables,
"query": query_str,
}
output_dict = get_template(query, access_token)["data"]["user"]["following"]
return RawFollows.model_validate(output_dict)
@@ -1,18 +0,0 @@
from typing import List, Optional
from pydantic import BaseModel, Field
from src.models import User
class PageInfo(BaseModel):
has_next_page: bool = Field(alias="hasNextPage")
end_cursor: Optional[str] = Field(alias="endCursor")
class Config:
allow_none = True
class RawFollows(BaseModel):
nodes: List[User]
page_info: PageInfo = Field(alias="pageInfo")
@@ -1,30 +0,0 @@
import json
import urllib.request
from typing import Any, Dict
BLACKLIST = [".md"]
with urllib.request.urlopen(
"https://raw.githubusercontent.com/blakeembrey/language-map/main/languages.json"
) as url:
data: Dict[str, Dict[str, Any]] = json.loads(url.read().decode())
languages = {
k: v
for k, v in data.items()
if v["type"] in ["programming", "markup"] and "color" in v and "extensions" in v
}
extensions: Dict[str, Dict[str, str]] = {}
for lang_name, lang in languages.items():
for extension in lang["extensions"]:
if extension not in BLACKLIST:
extensions[extension] = {"color": lang["color"], "name": lang_name}
extensions = dict(sorted(extensions.items(), key=lambda x: x[0]))
extensions[".tsx"]["name"] = "TypeScript"
extensions[".tsx"]["color"] = "#2B7489"
extensions[".cs"]["name"] = "C#"
extensions[".cs"]["color"] = "#178600"
extensions[".ml"]["name"] = "OCaml"
extensions[".ml"]["color"] = "#3BE133"
with open("src/data/github/extensions.json", "w") as f:
json.dump(extensions, f, indent=4)
@@ -1,17 +0,0 @@
from src.data.github.rest.commit import get_commit_files
from src.data.github.rest.models import RawCommit, RawCommitFile
from src.data.github.rest.repo import get_repo_commits, get_repo_stargazers
from src.data.github.rest.template import RESTError, RESTErrorNotFound
from src.data.github.rest.user import get_user, get_user_starred_repos
__all__ = [
"get_commit_files",
"RawCommit",
"RawCommitFile",
"get_repo_commits",
"get_repo_stargazers",
"RESTError",
"RESTErrorNotFound",
"get_user",
"get_user_starred_repos",
]
@@ -1,28 +0,0 @@
from typing import List, Optional
from src.data.github.rest.models import RawCommitFile
from src.data.github.rest.template import get_template
BASE_URL = "https://api.github.com/repos/"
def get_commit_files(
owner: str, repo: str, sha: str, access_token: Optional[str] = None
) -> Optional[List[RawCommitFile]]:
"""
Returns raw repository data
:param owner: repository owner
:param repo: repository name
:param sha: commit sha
:param access_token: GitHub access token
:return: repository data
"""
try:
output = get_template(
BASE_URL + owner + "/" + repo + "/commits/" + sha, access_token
)
files = output["files"]
return [RawCommitFile.model_validate(f) for f in files]
except Exception:
return None
@@ -1,14 +0,0 @@
from datetime import datetime
from pydantic import BaseModel
class RawCommit(BaseModel):
timestamp: datetime
node_id: str
class RawCommitFile(BaseModel):
filename: str
additions: int
deletions: int
@@ -1,174 +0,0 @@
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional
from src.data.github.rest.models import RawCommit
from src.data.github.rest.template import RESTError, get_template, get_template_plural
BASE_URL = "https://api.github.com/repos/"
# NOTE: unused, untested
def get_repo(access_token: str, owner: str, repo: str) -> Dict[str, Any]:
"""
Returns raw repository data
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:return: repository data
"""
return get_template(BASE_URL + owner + "/" + repo, access_token)
# NOTE: unused, untested
def get_repo_languages(
access_token: str, owner: str, repo: str
) -> List[Dict[str, Any]]:
"""
Returns repository language breakdown
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:return: repository language breakdown
"""
return get_template_plural(
BASE_URL + owner + "/" + repo + "/languages", access_token
)
def get_repo_stargazers(
access_token: str, owner: str, repo: str, per_page: int = 100, page: int = 1
) -> List[Dict[str, Any]]:
"""
Returns stargazers with timestamp for repository
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:param per_page: number of items per page
:param page: page number
:return: stargazers with timestamp for repository
"""
return get_template_plural(
BASE_URL + owner + "/" + repo + "/stargazers",
access_token,
per_page=per_page,
page=page,
accept_header="applicaiton/vnd.github.v3.star+json",
)
# NOTE: unused, untested
# does not accept per page, exceeds if necessary
def get_repo_code_frequency(access_token: str, owner: str, repo: str) -> Dict[str, Any]:
"""
Returns code frequency for repository
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:return: code frequency for repository
"""
return get_template(
BASE_URL + owner + "/" + repo + "/stats/code_frequency", access_token
)
# NOTE: unused, untested
def get_repo_commit_activity(
access_token: str, owner: str, repo: str
) -> Dict[str, Any]:
"""
Returns commit activity for past year, broken by week
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:return: commit activity for past year, broken by week
"""
return get_template(
BASE_URL + owner + "/" + repo + "/stats/commit_activity", access_token
)
# NOTE: unused, untested
def get_repo_contributors(access_token: str, owner: str, repo: str) -> Dict[str, Any]:
"""
Returns contributors for a repository
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:return: contributors for a repository
"""
return get_template(
BASE_URL + owner + "/" + repo + "/stats/contributors", access_token
)
# NOTE: unused, untested
def get_repo_weekly_commits(access_token: str, owner: str, repo: str) -> Dict[str, Any]:
"""
Returns contributions by week, owner/non-owner
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:return: contributions by week, owner/non-owner
"""
return get_template(
BASE_URL + owner + "/" + repo + "/stats/participation", access_token
)
# NOTE: unused, untested
def get_repo_hourly_commits(access_token: str, owner: str, repo: str) -> Dict[str, Any]:
"""
Returns contributions by day, hour for repository
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
"""
return get_template(
BASE_URL + owner + "/" + repo + "/stats/punch_card", access_token
)
def get_repo_commits(
owner: str,
repo: str,
user: Optional[str] = None,
since: Optional[datetime] = None,
until: Optional[datetime] = None,
page: int = 1,
access_token: Optional[str] = None,
) -> List[RawCommit]:
"""
Returns most recent commits
:param access_token: GitHub access token
:param owner: repository owner
:param repo: repository name
:param user: optional GitHub user if not owner
:param since: optional datetime to start from
:param until: optional datetime to end at
:param page: optional page number
:return: Up to 100 commits from page
"""
user = user if user is not None else owner
query = BASE_URL + owner + "/" + repo + "/commits?author=" + user
if since is not None:
query += f"&since={str(since)}"
if until is not None:
query += f"&until={str(until)}"
try:
data = get_template_plural(query, access_token, page=page)
def extract_info(x: Any) -> RawCommit:
dt = x["commit"]["committer"]["date"]
temp = {
"timestamp": datetime.strptime(dt, "%Y-%m-%dT%H:%M:%SZ"),
"node_id": x["node_id"],
}
return RawCommit.model_validate(temp)
return list(map(extract_info, data))
except RESTError:
return []
except Exception as e:
logging.exception(e)
return []
@@ -1,112 +0,0 @@
from datetime import datetime
from typing import Any, Dict, List, Optional
import requests
from requests.exceptions import ReadTimeout
from src.constants import TIMEOUT
from src.data.github.utils import get_access_token
s = requests.session()
class RESTError(Exception):
pass
class RESTErrorUnauthorized(RESTError):
pass
class RESTErrorNotFound(RESTError):
pass
class RESTErrorEmptyRepo(RESTError):
pass
class RESTErrorTimeout(RESTError):
pass
def _get_template(
query: str,
params: Dict[str, Any],
accept_header: str,
access_token: Optional[str] = None,
retries: int = 0,
) -> Any:
"""
Internal template for interacting with the GitHub REST API
:param query: The query to be sent to the GitHub API
:param params: The parameters to be sent to the GitHub API
:param access_token: The access token to be sent to the GitHub API
:param accept_header: The accept header to be sent to the GitHub API
:return: The response from the GitHub API
"""
start = datetime.now()
new_access_token = get_access_token(access_token)
headers: Dict[str, str] = {
"Accept": accept_header,
"Authorization": f"bearer {new_access_token}",
}
try:
r = s.get(query, params=params, headers=headers, timeout=TIMEOUT)
except ReadTimeout:
raise RESTErrorTimeout("REST Error: Request Timeout")
if r.status_code == 200:
print("REST API", new_access_token, datetime.now() - start)
return r.json()
if r.status_code == 401:
raise RESTErrorUnauthorized("REST Error: Unauthorized")
if r.status_code == 404:
raise RESTErrorNotFound("REST Error: Not Found")
if r.status_code == 409:
raise RESTErrorEmptyRepo("REST Error: Empty Repository")
if retries < 3:
print("REST Error, Retrying:", new_access_token)
return _get_template(query, params, accept_header, access_token, retries + 1)
raise RESTError(f"REST Error: {str(r.status_code)}")
def get_template(
query: str,
access_token: Optional[str] = None,
accept_header: str = "application/vnd.github.v3+json",
) -> Dict[str, Any]:
"""
Template for interacting with the GitHub REST API (singular)
:param query: The query to be sent to the GitHub API
:param access_token: The access token to be sent to the GitHub API
:param accept_header: The accept header to be sent to the GitHub API
:return: The response from the GitHub API
"""
return _get_template(query, {}, accept_header, access_token)
def get_template_plural(
query: str,
access_token: Optional[str] = None,
per_page: int = 100,
page: int = 1,
accept_header: str = "application/vnd.github.v3+json",
) -> List[Dict[str, Any]]:
"""
Template for interacting with the GitHub REST API (plural)
:param query: The query to be sent to the GitHub API
:param access_token: The access token to be sent to the GitHub API
:param per_page: The number of items to be returned per page
:param page: The page number to be returned
:param accept_header: The accept header to be sent to the GitHub API
:return: The response from the GitHub API
"""
params: Dict[str, str] = {"per_page": str(per_page), "page": str(page)}
return _get_template(query, params, accept_header, access_token)
@@ -1,32 +0,0 @@
from typing import Any, Dict, List
from src.data.github.rest.template import get_template, get_template_plural
BASE_URL = "https://api.github.com/users/"
def get_user(user_id: str, access_token: str) -> Dict[str, Any]:
"""
Returns raw user data
:param user_id: GitHub user id
:param access_token: GitHub access token
"""
return get_template(BASE_URL + user_id, access_token)
def get_user_starred_repos(
user_id: str, access_token: str, per_page: int = 100, page: int = 1
) -> List[Dict[str, Any]]:
"""
Returns list of starred repos
:param user_id: GitHub user id
:param access_token: GitHub access token
:param per_page: number of repos to return per page
"""
return get_template_plural(
BASE_URL + user_id + "/starred",
access_token,
per_page=per_page,
page=page,
accept_header="application/vnd.github.v3.star+json",
)
@@ -1,7 +0,0 @@
from typing import Optional
from src.data.mongo.secret import get_random_key
def get_access_token(access_token: Optional[str] = None) -> str:
return access_token if access_token is not None else get_random_key()
-29
View File
@@ -1,29 +0,0 @@
from motor.core import AgnosticCollection
from motor.motor_asyncio import AsyncIOMotorClient
from src.constants import LOCAL, MONGODB_PASSWORD, PROD
def get_conn_str(password: str, database: str) -> str:
return f"mongodb+srv://root:{password}@backend2.e50j8dp.mongodb.net/{database}?retryWrites=true&w=majority"
if LOCAL:
DB = None
elif PROD:
conn_str = get_conn_str(MONGODB_PASSWORD, "prod_backend")
CLIENT = AsyncIOMotorClient(
conn_str, serverSelectionTimeoutMS=5000, tlsInsecure=True
)
DB = CLIENT.prod_backend # type: ignore
else:
conn_str = get_conn_str(MONGODB_PASSWORD, "dev_backend")
CLIENT = AsyncIOMotorClient( # type: ignore
conn_str, serverSelectionTimeoutMS=5000, tlsInsecure=True
)
DB = CLIENT.dev_backend # type: ignore
# Overwrite type since only None if Local=True
SECRETS: AgnosticCollection = None if DB is None else DB.secrets # type: ignore
USERS: AgnosticCollection = None if DB is None else DB.users # type: ignore
USER_MONTHS: AgnosticCollection = None if DB is None else DB.user_months # type: ignore
@@ -1,3 +0,0 @@
from src.data.mongo.secret.functions import get_random_key, update_keys
__all__ = ["get_random_key", "update_keys"]
@@ -1,35 +0,0 @@
from datetime import timedelta
from random import randint
from typing import Any, Dict, List, Optional, Tuple
from src.constants import TEST_TOKEN
from src.data.mongo.main import SECRETS
from src.data.mongo.secret.models import SecretModel
from src.utils import alru_cache
@alru_cache(ttl=timedelta(minutes=15))
async def get_keys(no_cache: bool = False) -> Tuple[bool, List[str]]:
return (False, [])
secrets: Optional[Dict[str, Any]] = await SECRETS.find_one({"project": "main"})
if secrets is None:
return (False, [])
tokens = SecretModel.model_validate(secrets).access_tokens
return (True, tokens)
secret_keys: List[str] = []
async def update_keys(no_cache: bool = False) -> None:
global secret_keys
secret_keys = await get_keys(no_cache=no_cache)
def get_random_key() -> str:
global secret_keys
if len(secret_keys) == 0:
return TEST_TOKEN
return secret_keys[randint(0, len(secret_keys) - 1)]
@@ -1,8 +0,0 @@
from typing import List
from pydantic import BaseModel
class SecretModel(BaseModel):
project: str
access_tokens: List[str]
@@ -1,13 +0,0 @@
from src.data.mongo.user.functions import delete_user, is_user_key, update_user
from src.data.mongo.user.get import get_full_user, get_public_user
from src.data.mongo.user.models import FullUserModel, PublicUserModel
__all__ = [
"delete_user",
"is_user_key",
"update_user",
"get_full_user",
"get_public_user",
"FullUserModel",
"PublicUserModel",
]
@@ -1,28 +0,0 @@
from typing import Any, Dict, Optional
from src.data.mongo.main import USERS
async def is_user_key(user_id: str, user_key: str) -> bool:
user: Optional[dict[str, str]] = await USERS.find_one(
{"user_id": user_id}, {"user_key": 1}
)
return user is not None and user.get("user_key", "") == user_key
async def update_user(user_id: str, raw_user: Dict[str, Any]) -> None:
await USERS.update_one(
{"user_id": user_id},
{"$set": raw_user},
upsert=True,
)
async def delete_user(user_id: str, user_key: str, use_user_key: bool = True) -> bool:
if use_user_key:
is_key = await is_user_key(user_id, user_key)
if not is_key:
return False
await USERS.delete_one({"user_id": user_id})
return True
@@ -1,38 +0,0 @@
from typing import Any, Dict, Optional, Tuple
from pydantic import ValidationError
from src.data.mongo.main import USERS
from src.data.mongo.user.models import FullUserModel, PublicUserModel
from src.utils import alru_cache
@alru_cache()
async def get_public_user(
user_id: str, no_cache: bool = False
) -> Tuple[bool, Optional[PublicUserModel]]:
return (False, None)
user: Optional[Dict[str, Any]] = await USERS.find_one({"user_id": user_id})
if user is None:
# flag is false, don't cache
return (False, None)
try:
return (True, PublicUserModel.model_validate(user))
except (TypeError, KeyError, ValidationError):
return (False, None)
@alru_cache()
async def get_full_user(
user_id: str, no_cache: bool = False
) -> Tuple[bool, Optional[FullUserModel]]:
user: Optional[Dict[str, Any]] = await USERS.find_one({"user_id": user_id})
if user is None:
# flag is false, don't cache
return (False, None)
try:
return (True, FullUserModel.model_validate(user))
except (TypeError, KeyError, ValidationError):
return (False, None)
@@ -1,21 +0,0 @@
from typing import Optional
from pydantic import BaseModel, validator
class PublicUserModel(BaseModel):
user_id: str
access_token: str
private_access: Optional[bool]
class Config:
from_attributes = True
validate_assignment = True
@validator("private_access", pre=True, always=True)
def set_name(cls, private_access: Optional[bool]):
return False if private_access is None else private_access
class FullUserModel(PublicUserModel):
user_key: Optional[str]
@@ -1,5 +0,0 @@
from src.data.mongo.user_months.functions import set_user_month
from src.data.mongo.user_months.get import get_user_months
from src.data.mongo.user_months.models import UserMonth
__all__ = ["set_user_month", "get_user_months", "UserMonth"]
@@ -1,14 +0,0 @@
from src.data.mongo.main import USER_MONTHS
from src.data.mongo.user_months.models import UserMonth
async def set_user_month(user_month: UserMonth):
return
compressed_user_month = user_month.model_dump()
compressed_user_month["data"] = user_month.data.compress()
await USER_MONTHS.update_one(
{"user_id": user_month.user_id, "month": user_month.month},
{"$set": compressed_user_month},
upsert=True,
)
@@ -1,53 +0,0 @@
from datetime import date, datetime
from typing import Any, Dict, List
from src.constants import API_VERSION, USER_WHITELIST
from src.data.mongo.main import USER_MONTHS
from src.data.mongo.user_months.models import UserMonth
from src.models import UserPackage
async def get_user_months(
user_id: str, private_access: bool, start_month: date, end_month: date
) -> List[UserMonth]:
return []
start = datetime(start_month.year, start_month.month, 1)
end = datetime(end_month.year, end_month.month, 28)
today = datetime.now()
filters = {
"user_id": user_id,
"month": {"$gte": start, "$lte": end},
"version": API_VERSION,
}
if private_access:
filters["private"] = True
months: List[Dict[str, Any]] = await USER_MONTHS.find(filters).to_list(length=None) # type: ignore
months_data: List[UserMonth] = []
for month in months:
date_obj: datetime = month["month"]
complete = (
not (date_obj.year == today.year and date_obj.month == today.month)
or user_id in USER_WHITELIST
)
try:
data = UserPackage.decompress(month["data"])
months_data.append(
UserMonth.model_validate(
{
"user_id": user_id,
"month": month["month"],
"version": API_VERSION,
"private": month["private"],
"complete": complete,
"data": data.model_dump(),
}
)
)
except Exception:
pass
return months_data
@@ -1,14 +0,0 @@
from datetime import datetime
from pydantic import BaseModel
from src.models import UserPackage
class UserMonth(BaseModel):
user_id: str
month: datetime
version: float
private: bool
complete: bool
data: UserPackage
-70
View File
@@ -1,70 +0,0 @@
from typing import Dict
import sentry_sdk
from dotenv import find_dotenv, load_dotenv
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
load_dotenv(find_dotenv())
# flake8: noqa E402
# add endpoints here (after load dotenv)
from src.constants import PROD, SENTRY_DSN
from src.routers import (
asset_router,
auth_router,
dev_router,
user_router,
wrapped_router,
)
"""
SETUP
"""
app = FastAPI()
origins = [
"http://localhost:3000",
"http://localhost:3001",
"https://githubtrends.io",
"https://www.githubtrends.io",
"https://githubwrapped.io",
"https://www.githubwrapped.io",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
sentry_sdk.init(
SENTRY_DSN,
traces_sample_rate=(1.0 if PROD else 1.0),
)
app.add_middleware(
SentryAsgiMiddleware,
)
@app.get("/")
async def read_root() -> Dict[str, str]:
return {"Hello": "World"}
@app.get("/info")
def get_info() -> Dict[str, bool]:
return {"PROD": PROD}
app.include_router(asset_router, prefix="/assets", tags=["Assets"])
app.include_router(auth_router, prefix="/auth", tags=["Auth"])
app.include_router(dev_router, prefix="/dev", tags=["Dev"])
app.include_router(user_router, prefix="/user", tags=["Users"])
app.include_router(wrapped_router, prefix="/wrapped", tags=["Wrapped"])
-46
View File
@@ -1,46 +0,0 @@
from src.models.user.contribs import (
ContributionDay,
Language,
RepoContributionStats,
UserContributions,
)
from src.models.user.follows import User, UserFollows
from src.models.user.main import UserPackage
from src.models.wrapped.calendar import (
CalendarData,
CalendarDayDatum,
CalendarLanguageDayDatum,
)
from src.models.wrapped.langs import LangData, LangDatum
from src.models.wrapped.main import WrappedPackage
from src.models.wrapped.numeric import ContribStats, LOCStats, MiscStats, NumericData
from src.models.wrapped.repos import RepoData, RepoDatum
from src.models.wrapped.time import DayData, MonthData, TimeDatum
from src.models.wrapped.timestamps import TimestampData, TimestampDatum
__all__ = [
"ContributionDay",
"Language",
"RepoContributionStats",
"UserContributions",
"User",
"UserFollows",
"UserPackage",
"CalendarData",
"CalendarDayDatum",
"CalendarLanguageDayDatum",
"LangData",
"LangDatum",
"WrappedPackage",
"ContribStats",
"LOCStats",
"MiscStats",
"NumericData",
"RepoData",
"RepoDatum",
"DayData",
"MonthData",
"TimeDatum",
"TimestampData",
"TimestampDatum",
]
-12
View File
@@ -1,12 +0,0 @@
from datetime import date
from typing import Optional
from pydantic import BaseModel
class UpdateUserBackgroundTask(BaseModel):
user_id: str
access_token: Optional[str]
private_access: bool
start_date: Optional[date]
end_date: Optional[date]
-23
View File
@@ -1,23 +0,0 @@
from typing import List, Optional
from pydantic import BaseModel
class LanguageStats(BaseModel):
lang: str
loc: int
percent: float
color: Optional[str]
class RepoLanguage(BaseModel):
lang: str
color: Optional[str]
loc: int
class RepoStats(BaseModel):
repo: str
private: bool
langs: List[RepoLanguage]
loc: int
@@ -1,310 +0,0 @@
from datetime import date, datetime
from typing import Any, Dict, List, Optional, Tuple
from pydantic import BaseModel
class Language(BaseModel):
color: Optional[str]
additions: int
deletions: int
def compress(self) -> List[Any]:
return [self.color, self.additions, self.deletions]
@classmethod
def decompress(cls, data: List[Any]) -> "Language":
return Language(color=data[0], additions=data[1], deletions=data[2])
def __add__(self, other: "Language") -> "Language":
return Language(
color=self.color,
additions=self.additions + other.additions,
deletions=self.deletions + other.deletions,
)
class ContributionStats(BaseModel):
contribs_count: int
commits_count: int
issues_count: int
prs_count: int
reviews_count: int
repos_count: int
other_count: int
languages: Dict[str, Language]
def compress(self) -> List[Any]:
out: List[Any] = [
[
self.contribs_count,
self.commits_count,
self.issues_count,
self.prs_count,
self.reviews_count,
self.repos_count,
self.other_count,
],
*[[name] + stats.compress() for name, stats in self.languages.items()],
]
return out
@classmethod
def decompress(cls, data: List[Any]) -> "ContributionStats":
return ContributionStats(
contribs_count=data[0][0],
commits_count=data[0][1],
issues_count=data[0][2],
prs_count=data[0][3],
reviews_count=data[0][4],
repos_count=data[0][5],
other_count=data[0][6],
languages={x[0]: Language.decompress(x[1:]) for x in data[1:]},
)
def __add__(self, other: "ContributionStats") -> "ContributionStats":
languages = self.languages
for lang, lang_obj in other.languages.items():
if lang in languages:
languages[lang] += lang_obj
else:
languages[lang] = lang_obj
return ContributionStats(
contribs_count=self.contribs_count + other.contribs_count,
commits_count=self.commits_count + other.commits_count,
issues_count=self.issues_count + other.issues_count,
prs_count=self.prs_count + other.prs_count,
reviews_count=self.reviews_count + other.reviews_count,
repos_count=self.repos_count + other.repos_count,
other_count=self.other_count + other.other_count,
languages=languages,
)
@classmethod
def empty(cls) -> "ContributionStats":
return ContributionStats(
contribs_count=0,
commits_count=0,
issues_count=0,
prs_count=0,
reviews_count=0,
repos_count=0,
other_count=0,
languages={},
)
class ContributionLists(BaseModel):
commits: List[datetime]
issues: List[datetime]
prs: List[datetime]
reviews: List[datetime]
repos: List[datetime]
def compress(self) -> List[Any]:
return [self.commits, self.issues, self.prs, self.reviews, self.repos]
@classmethod
def decompress(cls, data: List[Any]) -> "ContributionLists":
return ContributionLists(
commits=data[0], issues=data[1], prs=data[2], reviews=data[3], repos=data[4]
)
class ContributionDay(BaseModel):
date: str
weekday: int
stats: ContributionStats
lists: ContributionLists
def compress(self) -> List[Any]:
return [
self.date,
self.weekday,
self.stats.compress(),
self.lists.compress(),
]
@classmethod
def decompress(cls, data: List[Any]) -> "ContributionDay":
return ContributionDay(
date=data[0],
weekday=data[1],
stats=ContributionStats.decompress(data[2]),
lists=ContributionLists.decompress(data[3]),
)
class RepoContributionStats(ContributionStats, BaseModel):
private: bool
contribs_count: int
commits_count: int
issues_count: int
prs_count: int
reviews_count: int
repos_count: int
other_count: int
languages: Dict[str, Language]
def compress(self) -> List[Any]:
out = super().compress()
out[0].append(self.private)
return out
@classmethod
def decompress(cls, data: List[Any]) -> "RepoContributionStats":
contribs = super().decompress(data).model_dump()
contribs["private"] = data[0][7]
return RepoContributionStats(**contribs)
def __add__( # type: ignore
self, other: "RepoContributionStats"
) -> "RepoContributionStats":
new_self = ContributionStats(**self.model_dump())
new_other = ContributionStats(**other.model_dump())
combined = (new_self + new_other).model_dump()
combined["private"] = self.private
return RepoContributionStats(**combined)
class UserContributions(BaseModel):
total_stats: ContributionStats
public_stats: ContributionStats
total: List[ContributionDay]
public: List[ContributionDay]
repo_stats: Dict[str, RepoContributionStats]
repos: Dict[str, List[ContributionDay]]
def compress(self) -> List[Any]:
new_total_stats = self.total_stats.compress()
new_public_stats = self.public_stats.compress()
new_total = [x.compress() for x in self.total]
new_public = [x.compress() for x in self.public]
new_repo_stats = {k: v.compress() for k, v in self.repo_stats.items()}
new_repos = {k: [x.compress() for x in v] for k, v in self.repos.items()}
return [
new_total_stats,
new_public_stats,
new_total,
new_public,
new_repo_stats,
new_repos,
]
@classmethod
def decompress(cls, data: List[Any]) -> "UserContributions":
total_stats = ContributionStats.decompress(data[0])
public_stats = ContributionStats.decompress(data[1])
total = [ContributionDay.decompress(x) for x in data[2]]
public = [ContributionDay.decompress(x) for x in data[3]]
repo_stats = {
k: RepoContributionStats.decompress(v) for k, v in data[4].items()
}
repos = {
k: [ContributionDay.decompress(x) for x in v] for k, v in data[5].items()
}
return UserContributions(
total_stats=total_stats,
public_stats=public_stats,
total=total,
public=public,
repo_stats=repo_stats,
repos=repos,
)
def __add__(self, other: "UserContributions") -> "UserContributions":
new_total_stats = self.total_stats + other.total_stats
new_public_stats = self.public_stats + other.public_stats
new_total = sorted(self.total + other.total, key=lambda x: x.date)
new_public = sorted(self.public + other.public, key=lambda x: x.date)
new_repo_stats = self.repo_stats
for repo, stats in other.repo_stats.items():
if repo in new_repo_stats:
new_repo_stats[repo] += stats
else:
new_repo_stats[repo] = stats
new_repos = self.repos
for repo, days in other.repos.items():
if repo in new_repos:
new_repos[repo] = sorted(new_repos[repo] + days, key=lambda x: x.date)
else:
new_repos[repo] = days
return UserContributions(
total_stats=new_total_stats,
public_stats=new_public_stats,
total=new_total,
public=new_public,
repo_stats=new_repo_stats,
repos=new_repos,
)
@staticmethod
def trim_contribs(
contribs: List[ContributionDay], start_date: date, end_date: date
) -> Tuple[List[ContributionDay], ContributionStats]:
new_total: List[ContributionDay] = []
for day in contribs:
curr_date = datetime.strptime(day.date, "%Y-%m-%d").date()
if curr_date >= start_date and curr_date <= end_date:
new_total.append(day)
new_languages: Dict[str, Language] = {}
for day in new_total:
for lang in day.stats.languages:
if lang in new_languages:
new_languages[lang] += day.stats.languages[lang]
else:
new_languages[lang] = day.stats.languages[lang]
new_stats = ContributionStats(
contribs_count=sum(x.stats.contribs_count for x in new_total),
commits_count=sum(x.stats.commits_count for x in new_total),
issues_count=sum(x.stats.issues_count for x in new_total),
prs_count=sum(x.stats.prs_count for x in new_total),
reviews_count=sum(x.stats.reviews_count for x in new_total),
repos_count=sum(x.stats.repos_count for x in new_total),
other_count=sum(x.stats.other_count for x in new_total),
languages=new_languages,
)
return new_total, new_stats
def trim(self, start: date, end: date) -> "UserContributions":
new_total, new_total_stats = self.trim_contribs(self.total, start, end)
new_public, new_public_stats = self.trim_contribs(self.public, start, end)
new_repos_dict: Dict[str, List[ContributionDay]] = {}
new_repo_stats_dict: Dict[str, RepoContributionStats] = {}
for repo_name, repo in self.repos.items():
new_repo_total, _new_repo_stats = self.trim_contribs(repo, start, end)
if len(new_repo_total) > 0:
new_repos_dict[repo_name] = new_repo_total
raw_new_repo_stats = _new_repo_stats.model_dump()
raw_new_repo_stats["private"] = self.repo_stats[repo_name].private
new_repo_stats = RepoContributionStats(**raw_new_repo_stats)
new_repo_stats_dict[repo_name] = new_repo_stats
return UserContributions(
total_stats=new_total_stats,
public_stats=new_public_stats,
total=new_total,
public=new_public,
repo_stats=new_repo_stats_dict,
repos=new_repos_dict,
)
@classmethod
def empty(cls) -> "UserContributions":
return UserContributions(
total_stats=ContributionStats.empty(),
public_stats=ContributionStats.empty(),
total=[],
public=[],
repo_stats={},
repos={},
)
@@ -1,17 +0,0 @@
from typing import List, Optional
from pydantic import BaseModel
class User(BaseModel):
name: Optional[str]
login: str
url: str
class Config:
allow_none = True
class UserFollows(BaseModel):
followers: List[User]
following: List[User]
-34
View File
@@ -1,34 +0,0 @@
from datetime import date
from typing import Any, Dict
from pydantic import BaseModel
from src.models.user.contribs import UserContributions
# from src.models.user.follows import UserFollows
class UserPackage(BaseModel):
contribs: UserContributions
incomplete: bool = False
def compress(self):
return {
"c": self.contribs.compress(),
}
@classmethod
def decompress(cls, data: Dict[str, Any]) -> "UserPackage":
return UserPackage(
contribs=UserContributions.decompress(data["c"]),
)
def __add__(self, other: "UserPackage") -> "UserPackage":
return UserPackage(contribs=self.contribs + other.contribs)
def trim(self, start: date, end: date) -> "UserPackage":
return UserPackage(contribs=self.contribs.trim(start, end))
@classmethod
def empty(cls) -> "UserPackage":
return UserPackage(contribs=UserContributions.empty())
@@ -1,28 +0,0 @@
from typing import Dict, List
from pydantic import BaseModel
class CalendarLanguageDayDatum(BaseModel):
loc_added: int
loc_changed: int
class CalendarDayDatum(BaseModel):
day: str
contribs: int
commits: int
issues: int
prs: int
reviews: int
loc_added: int
loc_changed: int
top_langs: Dict[str, CalendarLanguageDayDatum]
class CalendarData(BaseModel):
days: List[CalendarDayDatum]
@classmethod
def empty(cls) -> "CalendarData":
return CalendarData(days=[])
@@ -1,20 +0,0 @@
from typing import List
from pydantic import BaseModel
class LangDatum(BaseModel):
id: str
label: str
value: int
formatted_value: str
color: str
class LangData(BaseModel):
langs_changed: List[LangDatum]
langs_added: List[LangDatum]
@classmethod
def empty(cls) -> "LangData":
return LangData(langs_changed=[], langs_added=[])
@@ -1,31 +0,0 @@
from pydantic import BaseModel
from src.models.wrapped.calendar import CalendarData
from src.models.wrapped.langs import LangData
from src.models.wrapped.numeric import NumericData
from src.models.wrapped.repos import RepoData
from src.models.wrapped.time import DayData, MonthData
from src.models.wrapped.timestamps import TimestampData
class WrappedPackage(BaseModel):
month_data: MonthData
day_data: DayData
calendar_data: CalendarData
numeric_data: NumericData
repo_data: RepoData
lang_data: LangData
timestamp_data: TimestampData
incomplete: bool = False
@classmethod
def empty(cls) -> "WrappedPackage":
return WrappedPackage(
month_data=MonthData.empty(),
day_data=DayData.empty(),
calendar_data=CalendarData.empty(),
numeric_data=NumericData.empty(),
repo_data=RepoData.empty(),
lang_data=LangData.empty(),
timestamp_data=TimestampData.empty(),
)
@@ -1,85 +0,0 @@
from typing import Optional, Tuple
from pydantic import BaseModel
class ContribStats(BaseModel):
contribs: int
commits: int
issues: int
prs: int
reviews: int
other: int
@classmethod
def empty(cls) -> "ContribStats":
return ContribStats(
contribs=0,
commits=0,
issues=0,
prs=0,
reviews=0,
other=0,
)
class MiscStats(BaseModel):
total_days: int
longest_streak: int
longest_streak_days: Tuple[int, int, str, str]
longest_gap: int
longest_gap_days: Tuple[int, int, str, str]
weekend_percent: int
best_day_count: int
best_day_date: Optional[str]
best_day_index: Optional[int]
@classmethod
def empty(cls) -> "MiscStats":
return MiscStats(
total_days=0,
longest_streak=0,
longest_streak_days=(0, 0, "", ""),
longest_gap=0,
longest_gap_days=(0, 0, "", ""),
weekend_percent=0,
best_day_count=0,
best_day_date=None,
best_day_index=None,
)
class LOCStats(BaseModel):
loc_additions: str
loc_deletions: str
loc_changed: str
loc_added: str
loc_additions_per_commit: int
loc_deletions_per_commit: int
loc_changed_per_day: int
@classmethod
def empty(cls) -> "LOCStats":
return LOCStats(
loc_additions="0",
loc_deletions="0",
loc_changed="0",
loc_added="0",
loc_additions_per_commit=0,
loc_deletions_per_commit=0,
loc_changed_per_day=0,
)
class NumericData(BaseModel):
contribs: ContribStats
misc: MiscStats
loc: LOCStats
@classmethod
def empty(cls) -> "NumericData":
return NumericData(
contribs=ContribStats.empty(),
misc=MiscStats.empty(),
loc=LOCStats.empty(),
)
@@ -1,19 +0,0 @@
from typing import List
from pydantic import BaseModel
class RepoDatum(BaseModel):
id: int
label: str
value: int
formatted_value: str
class RepoData(BaseModel):
repos_changed: List[RepoDatum]
repos_added: List[RepoDatum]
@classmethod
def empty(cls) -> "RepoData":
return RepoData(repos_changed=[], repos_added=[])
@@ -1,26 +0,0 @@
from typing import List
from pydantic import BaseModel
class TimeDatum(BaseModel):
index: int
contribs: int
loc_changed: int
formatted_loc_changed: str
class MonthData(BaseModel):
months: List[TimeDatum]
@classmethod
def empty(cls) -> "MonthData":
return MonthData(months=[])
class DayData(BaseModel):
days: List[TimeDatum]
@classmethod
def empty(cls) -> "DayData":
return DayData(days=[])
@@ -1,17 +0,0 @@
from typing import List
from pydantic import BaseModel
class TimestampDatum(BaseModel):
type: str
weekday: int
timestamp: int
class TimestampData(BaseModel):
contribs: List[TimestampDatum]
@classmethod
def empty(cls) -> "TimestampData":
return TimestampData(contribs=[])
-67
View File
@@ -1,67 +0,0 @@
from typing import Any, Dict, Optional, Tuple
from src.data.github.auth import authenticate as github_authenticate
from src.data.mongo.user import (
PublicUserModel,
delete_user as db_delete_user,
get_public_user as db_get_public_user,
update_user as db_update_user,
)
from src.models.background import UpdateUserBackgroundTask
# frontend first calls set_user_key with code and user_key
# next they call authenticate which determines the user_id to associate with the code/user_key
# these actions should happen sequentially, so in-memory storage is fine
code_key_map: Dict[str, str] = {}
async def set_user_key(code: str, user_key: str) -> str:
code_key_map[code] = user_key
return user_key
async def authenticate(
code: str, private_access: bool
) -> Tuple[str, Optional[UpdateUserBackgroundTask]]:
user_id, access_token = await github_authenticate(code)
curr_user: Optional[PublicUserModel] = await db_get_public_user(user_id)
raw_user: Dict[str, Any] = {
"user_id": user_id,
"access_token": access_token,
"user_key": code_key_map.get(code, None),
"private_access": private_access,
}
background_task = None
if curr_user is not None:
curr_private_access = curr_user.private_access
new_private_access = curr_private_access or private_access
raw_user["private_access"] = new_private_access
if new_private_access != curr_private_access:
# new private access status
background_task = UpdateUserBackgroundTask(
user_id=user_id,
access_token=access_token,
private_access=new_private_access,
start_date=None,
end_date=None,
)
else:
# first time sign up
background_task = UpdateUserBackgroundTask(
user_id=user_id,
access_token=access_token,
private_access=private_access,
start_date=None,
end_date=None,
)
# await db_update_user(user_id, raw_user)
return user_id, background_task
async def delete_user(user_id: str, user_key: str, use_user_key: bool = True) -> bool:
return await db_delete_user(user_id, user_key, use_user_key)
@@ -1,4 +0,0 @@
from src.processing.user.commits import get_top_languages, get_top_repos
from src.processing.user.svg import svg_base
__all__ = ["get_top_languages", "get_top_repos", "svg_base"]
@@ -1,136 +0,0 @@
from typing import Any, Dict, List, Optional, Tuple, Union
from src.constants import DEFAULT_COLOR
from src.models import UserPackage
from src.models.svg import LanguageStats, RepoStats
dict_type = Dict[str, Union[str, int, float]]
def loc_metric_func(loc_metric: str, additions: int, deletions: int) -> int:
if loc_metric == "changed":
return additions + deletions
return additions - deletions
def get_top_languages(
data: UserPackage, loc_metric: str, include_private: bool
) -> Tuple[List[LanguageStats], int]:
raw_languages = (
data.contribs.total_stats.languages
if include_private
else data.contribs.public_stats.languages
)
languages_list = [
LanguageStats(
lang=lang,
color=stats.color or DEFAULT_COLOR,
loc=loc_metric_func(loc_metric, stats.additions, stats.deletions),
percent=-1,
)
for lang, stats in raw_languages.items()
]
languages_list = list(filter(lambda x: x.loc > 0, languages_list))
total_loc = sum(x.loc for x in languages_list) + 1
total = LanguageStats(lang="Total", color=None, loc=total_loc, percent=100)
languages_list = sorted(languages_list, key=lambda x: x.loc, reverse=True)
other = LanguageStats(lang="Other", color="#ededed", loc=0, percent=-1)
for language in languages_list[4:]:
other.loc = other.loc + language.loc
languages_list = [total] + languages_list[:4] + [other]
new_languages_list: List[LanguageStats] = []
for lang in languages_list:
lang.percent = float(round(100 * lang.loc / total_loc, 2))
if lang.percent > 1: # 1% minimum to show
new_languages_list.append(LanguageStats.model_validate(lang))
commits_excluded = data.contribs.public_stats.other_count
if include_private:
commits_excluded = data.contribs.total_stats.other_count
return new_languages_list, commits_excluded
def get_top_repos(
data: UserPackage, loc_metric: str, include_private: bool, group: str
) -> Tuple[List[RepoStats], int]:
repos: List[Any] = [
{
"repo": repo,
"private": repo_stats.private,
"langs": [
{
"lang": x[0],
"color": x[1].color,
"loc": loc_metric_func(loc_metric, x[1].additions, x[1].deletions),
}
for x in list(repo_stats.languages.items())
],
}
for repo, repo_stats in data.contribs.repo_stats.items()
if include_private or not repo_stats.private
]
for repo in repos:
repo["loc"] = sum(x["loc"] for x in repo["langs"]) # first estimate
repos = list(filter(lambda x: x["loc"] > 0, repos))
for repo in repos:
repo["langs"] = [x for x in repo["langs"] if x["loc"] > 0.05 * repo["loc"]]
repo["loc"] = sum(x["loc"] for x in repo["langs"]) # final estimate
repos = sorted(repos, key=lambda x: x["loc"], reverse=True)
new_repos = [
RepoStats.model_validate(x) for x in repos if x["loc"] > 0.01 * repos[0]["loc"]
]
commits_excluded = data.contribs.public_stats.other_count
if include_private:
commits_excluded = data.contribs.total_stats.other_count
# With n bars, group from n onwards into the last bar
bars = 4 # TODO: make this configurable (see issues)
if group == "none" or len(new_repos) <= bars:
return new_repos[:bars], commits_excluded
out_repos = []
other_repos = []
if group == "other":
out_repos = new_repos[: bars - 1]
other_repos = new_repos[bars - 1 :]
elif group == "private":
public_repos = [x for x in new_repos if not x.private]
private_repos = [x for x in new_repos if x.private]
if len(public_repos) < 4 and len(private_repos) > 0:
public_repos += private_repos[: bars - len(public_repos) - 1]
private_repos = private_repos[bars - len(public_repos) - 1 :]
out_repos = sorted(public_repos[: bars - 1], key=lambda x: x.loc, reverse=True)
other_repos = public_repos[bars - 1 :] + private_repos
else:
raise ValueError("Invalid group value")
other: Dict[str, Tuple[int, Optional[str]]] = {}
for repo in other_repos:
for _lang in repo.langs:
lang = _lang.lang
if lang not in other:
other[lang] = (0, _lang.color)
other[lang] = (other[lang][0] + _lang.loc, other[lang][1])
out_repos.append(
RepoStats(
repo="other/repos",
private=False,
langs=[{"lang": k, "loc": v[0], "color": v[1]} for k, v in other.items()], # type: ignore
loc=sum(v[0] for v in other.values()),
)
)
return out_repos, commits_excluded
@@ -1,32 +0,0 @@
from datetime import date
from typing import Optional, Tuple
from src.aggregation.layer2.user import get_user, get_user_demo
from src.models import UserPackage
from src.models.background import UpdateUserBackgroundTask
from src.utils import use_time_range
async def svg_base(
user_id: str,
start_date: date,
end_date: date,
time_range: str,
demo: bool,
no_cache: bool = False,
) -> Tuple[Optional[UserPackage], bool, Optional[UpdateUserBackgroundTask], str]:
# process time_range, start_date, end_date
time_range = "one_month" if demo else time_range
start_date, end_date, time_str = use_time_range(time_range, start_date, end_date)
complete = True # overwritten later if not complete
background_task = None
# fetch data, either using demo or user method
if demo:
output = await get_user_demo(user_id, start_date, end_date, no_cache=no_cache)
else:
output, complete, background_task = await get_user(
user_id, start_date, end_date, no_cache=no_cache
)
return output, complete, background_task, time_str
@@ -1,3 +0,0 @@
from src.processing.wrapped.main import query_wrapped_user
__all__ = ["query_wrapped_user"]
@@ -1,51 +0,0 @@
from datetime import datetime, timedelta
from typing import Any, Dict, List
from src.models import CalendarData, CalendarDayDatum, UserPackage
def get_calendar_data(data: UserPackage, year: int) -> CalendarData:
top_langs = [
x[0]
for x in sorted(
data.contribs.total_stats.languages.items(),
key=lambda x: x[1].additions + x[1].deletions,
reverse=True,
)[:5]
]
total_out: List[CalendarDayDatum] = []
items_dict = {item.date: item for item in data.contribs.total}
for i in range(365):
date = (datetime(year, 1, 1) + timedelta(days=i - 1)).strftime("%Y-%m-%d")
item = items_dict.get(date)
out: Dict[str, Any] = {
"day": date,
"contribs": 0,
"commits": 0,
"issues": 0,
"prs": 0,
"reviews": 0,
"loc_added": 0,
"loc_changed": 0,
"top_langs": {k: {"loc_added": 0, "loc_changed": 0} for k in top_langs},
}
if item is not None:
out["contribs"] = item.stats.contribs_count
out["commits"] = item.stats.commits_count
out["issues"] = item.stats.issues_count
out["prs"] = item.stats.prs_count
out["reviews"] = item.stats.reviews_count
for k, v in item.stats.languages.items():
if k in top_langs:
out["top_langs"][k]["loc_added"] = v.additions - v.deletions
out["top_langs"][k]["loc_changed"] = v.additions + v.deletions
out["loc_added"] += v.additions - v.deletions
out["loc_changed"] += v.additions + v.deletions
out_obj = CalendarDayDatum.model_validate(out)
total_out.append(out_obj)
return CalendarData.model_validate({"days": total_out})
@@ -1,47 +0,0 @@
from typing import List
from src.constants import DEFAULT_COLOR
from src.models import LangData, LangDatum, Language, UserPackage
from src.utils import format_number
def _count_loc(x: Language, metric: str) -> int:
if metric == "changed":
return x.additions + x.deletions
return x.additions - x.deletions
def get_lang_data(data: UserPackage) -> LangData:
out = {}
for m in ["changed", "added"]:
langs = sorted(
data.contribs.total_stats.languages.items(),
key=lambda x: _count_loc(x[1], m),
reverse=True,
)
lang_objs: List[LangDatum] = []
for k, v in list(langs)[:5]:
lang_data = {
"id": k,
"label": k,
"value": _count_loc(v, m),
"formatted_value": format_number(_count_loc(v, m)),
"color": v.color,
}
lang_objs.append(LangDatum.model_validate(lang_data))
# remaining languages
total_count = sum(_count_loc(v, m) for _, v in list(langs)[5:])
lang_data = {
"id": "other",
"label": "other",
"value": total_count,
"formatted_value": format_number(total_count),
"color": DEFAULT_COLOR,
}
if total_count > 100:
lang_objs.append(LangDatum.model_validate(lang_data))
out[f"langs_{m}"] = lang_objs
return LangData.model_validate(out)
@@ -1,34 +0,0 @@
from datetime import date, timedelta
from typing import Optional, Tuple
from src.aggregation.layer1 import query_user
from src.data.mongo.user import PublicUserModel, get_public_user as db_get_public_user
from src.models import UserPackage, WrappedPackage
from src.processing.wrapped.package import get_wrapped_data
from src.utils import alru_cache
@alru_cache(ttl=timedelta(hours=12))
async def query_wrapped_user(
user_id: str, year: int, no_cache: bool = False
) -> Tuple[bool, Optional[WrappedPackage]]:
start_date, end_date = date(year, 1, 1), date(year, 12, 31)
user: Optional[PublicUserModel] = await db_get_public_user(user_id)
private_access = False
access_token = None
if user is not None and user.private_access:
private_access = True
access_token = user.access_token
user_package: UserPackage = await query_user(
user_id,
access_token,
private_access,
start_date,
end_date,
max_time=40,
no_cache=True,
)
wrapped_package = get_wrapped_data(user_package, year)
# Don't cache if incomplete
return (not wrapped_package.incomplete, wrapped_package)
@@ -1,129 +0,0 @@
from collections import defaultdict
from datetime import datetime
from typing import Dict
from src.models import ContribStats, LOCStats, MiscStats, NumericData, UserPackage
def get_contrib_stats(data: UserPackage) -> ContribStats:
return ContribStats.model_validate(
{
"contribs": data.contribs.total_stats.contribs_count,
"commits": data.contribs.total_stats.commits_count,
"issues": data.contribs.total_stats.issues_count,
"prs": data.contribs.total_stats.prs_count,
"reviews": data.contribs.total_stats.reviews_count,
"other": data.contribs.total_stats.other_count,
}
)
def get_misc_stats(data: UserPackage, year: int) -> MiscStats:
weekdays: Dict[int, int] = defaultdict(int)
yeardays, distinct_days, total_contribs = {}, 0, 0
for item in data.contribs.total:
count = item.stats.contribs_count
weekdays[item.weekday] += count
total_contribs += item.stats.contribs_count
if count > 0:
date = datetime.fromisoformat(item.date)
yeardays[date.timetuple().tm_yday - 1] = 1
distinct_days += 1
curr, best, best_dates = 0, 0, (1, 1)
for i in range(366):
curr = curr + 1 if i in yeardays else 0
if curr > best:
best = curr
best_dates = (i - curr + 2, i + 1)
longest_streak = max(best, curr)
longest_streak_days = (
best_dates[0],
best_dates[1],
datetime.fromordinal(max(1, best_dates[0])).strftime("%b %d"),
datetime.fromordinal(max(1, best_dates[1])).strftime("%b %d"),
)
curr, best, best_dates = 0, 0, (1, 1)
days = (datetime.now() - datetime(year, 1, 1)).days
for i in range(min(days, 365)):
curr = 0 if i in yeardays else curr + 1
if curr > best:
best = curr
best_dates = (i - curr + 2, i + 1)
longest_gap = max(best, curr)
longest_gap_days = (
best_dates[0],
best_dates[1],
datetime.fromordinal(max(1, best_dates[0])).strftime("%b %d"),
datetime.fromordinal(max(1, best_dates[1])).strftime("%b %d"),
)
weekend_percent = (weekdays[0] + weekdays[6]) / max(1, total_contribs)
best_day_count, best_day_date, best_day_index = 0, None, None
if len(data.contribs.total) > 0:
best_day = max(data.contribs.total, key=lambda x: x.stats.contribs_count)
best_day_index = datetime.fromisoformat(best_day.date).timetuple().tm_yday
best_day_count = best_day.stats.contribs_count
best_day_date = best_day.date
return MiscStats.model_validate(
{
"total_days": distinct_days,
"longest_streak": longest_streak,
"longest_streak_days": longest_streak_days,
"longest_gap": longest_gap,
"longest_gap_days": longest_gap_days,
"weekend_percent": round(100 * weekend_percent),
"best_day_count": best_day_count,
"best_day_date": best_day_date,
"best_day_index": best_day_index,
}
)
def format_loc_number(number: int) -> str:
if number < 1e3:
return str(100 * round(number / 100))
if number < 1e6:
return f"{str(round(number / 1000.0))},000"
return f"{str(round(number / 1000000.0))},000,000"
def get_loc_stats(data: UserPackage) -> LOCStats:
dataset = data.contribs.total_stats.languages.values()
return LOCStats.model_validate(
{
"loc_additions": format_loc_number(sum(x.additions for x in dataset)),
"loc_deletions": format_loc_number(sum(x.deletions for x in dataset)),
"loc_changed": format_loc_number(
sum(x.additions + x.deletions for x in dataset)
),
"loc_added": format_loc_number(
sum(x.additions - x.deletions for x in dataset)
),
"loc_additions_per_commit": round(
(
sum(x.additions for x in dataset)
/ max(1, data.contribs.total_stats.commits_count)
)
),
"loc_deletions_per_commit": round(
(
sum(x.deletions for x in dataset)
/ max(1, data.contribs.total_stats.commits_count)
)
),
"loc_changed_per_day": round(
sum(x.additions + x.deletions for x in dataset) / 365
),
}
)
def get_numeric_data(data: UserPackage, year: int) -> NumericData:
return NumericData.model_validate(
{
"contribs": get_contrib_stats(data),
"misc": get_misc_stats(data, year),
"loc": get_loc_stats(data),
}
)
@@ -1,32 +0,0 @@
from src.models import UserPackage, WrappedPackage
from src.processing.wrapped.calendar import get_calendar_data
from src.processing.wrapped.langs import get_lang_data
from src.processing.wrapped.numeric import get_numeric_data
from src.processing.wrapped.repos import get_repo_data
from src.processing.wrapped.time import get_day_data, get_month_data
from src.processing.wrapped.timestamps import get_timestamp_data
# from src.processing.user.follows import get_user_follows
def get_wrapped_data(user_package: UserPackage, year: int) -> WrappedPackage:
"""packages all processing steps for the user query"""
month_data = get_month_data(user_package)
day_data = get_day_data(user_package)
calendar_data = get_calendar_data(user_package, year)
numeric_data = get_numeric_data(user_package, year)
repo_data = get_repo_data(user_package)
lang_data = get_lang_data(user_package)
timestamp_data = get_timestamp_data(user_package)
return WrappedPackage(
month_data=month_data,
day_data=day_data,
calendar_data=calendar_data,
numeric_data=numeric_data,
repo_data=repo_data,
lang_data=lang_data,
timestamp_data=timestamp_data,
incomplete=user_package.incomplete,
)

Some files were not shown because too many files have changed in this diff Show More