Add clone, referring, and traffic sqlmodels

Author: astrochun

github_stats_pages/db.py

@@ -0,0 +1,79 @@
+from functools import partial
+from pathlib import Path
+from typing import Union, Type
+
+import pandas as pd
+from sqlalchemy.future import Engine
+from sqlalchemy.exc import NoResultFound
+from sqlmodel import SQLModel, Session, create_engine, select
+
+from .models import Clone, Referring, Traffic, Paths
+from .logger import app_log as log
+
+SQLITE_FILE_NAME = "data/sqlite3.db"
+
+
+def configure(test: bool = False, echo: bool = False) -> Engine:
+    sqlite_file_name = "test_data/sqlite3.db" if test else SQLITE_FILE_NAME
+    sqlite_url = f"sqlite:///{sqlite_file_name}"
+    return create_engine(sqlite_url, echo=echo)
+
+
+def create_db_and_tables(test: bool = False):
+    engine = configure(test=test, echo=True)
+    SQLModel.metadata.create_all(engine)
+    return engine
+
+
+def migrate_csv(
+    filename: Union[str, Path],
+    model: Type[SQLModel],
+    engine: Engine,
+    skip_rows: Union[int, None] = None,
+):
+    """Migrate CSV over to SQLite"""
+
+    names = list(
+        map(
+            lambda f: f.name,
+            filter(lambda x: x.required, model.__fields__.values()),
+        )
+    )
+    log.info(f"[yellow]Loading: {filename}")
+    df = pd.read_csv(filename, header=None, skiprows=skip_rows, names=names)
+    func = partial(query, engine=engine, model=model)
+
+    query_results = list(map(func, df["repository_name"], df["date"]))
+    new_df: pd.DataFrame = df.iloc[
+        [idx for idx, item in enumerate(query_results) if not item]
+    ]
+    if new_df.empty:
+        log.info("No new records!")
+    else:
+        log.info(f"New records found: {len(new_df)}")
+        log.info("[bold yellow]Adding data")
+        new_df.to_sql(
+            model.__name__.lower(), engine, if_exists="append", index=False
+        )
+
+        if len(new_df) < len(df):
+            log.info("[orange]Some records exists in db")
+
+
+def query(
+    repository_name: str,
+    date: str,
+    engine: Engine,
+    model: Union[SQLModel, Clone, Referring, Paths, Traffic],
+) -> Union[SQLModel, Clone, Referring, Paths, Traffic, None]:
+
+    with Session(engine) as session:
+        result = session.exec(
+            select(model).where(
+                model.repository_name == repository_name, model.date == date
+            )
+        )
+        try:
+            return result.one()
+        except NoResultFound:
+            return

github_stats_pages/models/__init__.py

@@ -1,3 +1,4 @@
 from .clone import Clone  # noqa: F401
 from .referring import Referring  # noqa: F401
+from .paths import Paths  # noqa: F401
 from .traffic import Traffic  # noqa: F401

github_stats_pages/models/paths.py

@@ -0,0 +1,12 @@
+from typing import Optional
+
+from sqlmodel import SQLModel, Field
+
+
+class Paths(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    date: str
+    path: str
+    title: str
+    views: int
+    unique_views: int