feat: Built a github repository tracker that supports vector search

hackathon_project_tracker/.gitignore

@@ -0,0 +1,5 @@
+*.db
+*.py[cod]
+.web
+__pycache__/
+assets/external/

hackathon_project_tracker/hackathon_project_tracker/app_style.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import reflex as rx
+
+
+@dataclass
+class Style:
+    """The style for the app."""
+    dark: dict[str, str | rx.Component] = field(
+        default_factory=lambda: {
+            "background_color": "#121212",
+        },
+    )

hackathon_project_tracker/hackathon_project_tracker/hackathon_project_tracker.py

@@ -0,0 +1,22 @@
+# trunk-ignore-all(trunk/ignore-does-nothing)
+import reflex as rx
+
+from .app_style import Style as AppStyle
+from .pages.project_tracker.page import index
+from .pages.project_tracker.state import State
+
+APP_STYLE: AppStyle = AppStyle()
+
+app = rx.App(
+    style=APP_STYLE.dark,
+    theme=rx.theme(
+        has_background=True,
+        radius="large",
+        accent_color="teal",
+    ),
+)
+app.add_page(
+    component=index,
+    route="/",
+    on_load=State.on_load, # trunk-ignore(pyright/reportArgumentType)
+)

hackathon_project_tracker/hackathon_project_tracker/helper_chroma/__init__.py

@@ -0,0 +1,4 @@
+__all__ = ["add_item", "get_items", "set_up_client_from_tokens"]
+
+from .helper_chroma import add_item, get_items, set_up_client_from_tokens
+

hackathon_project_tracker/hackathon_project_tracker/helper_chroma/helper_chroma.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import chromadb
+
+from hackathon_project_tracker.otel import tracer
+
+if TYPE_CHECKING:
+    from hackathon_project_tracker.models.project import Project
+
+
+SPAN_KEY: str = "chroma"
+
+
+def set_up_client_from_tokens(
+    tokens: dict[str, str],
+) -> chromadb.api.ClientAPI:
+    required_tokens: list[str] = ["CHROMA_TENANT", "CHROMA_DATABASE", "CHROMA_API_KEY"]
+    missing_tokens = [token for token in required_tokens if not tokens.get(token)]
+    if missing_tokens:
+        error_msg: str = f"Missing required tokens: {', '.join(missing_tokens)}"
+        raise ValueError(error_msg)
+
+    return chromadb.HttpClient(
+        ssl=True,
+        host="api.trychroma.com",
+        tenant=str(tokens.get("CHROMA_TENANT")),
+        database=str(tokens.get("CHROMA_DATABASE")),
+        headers={
+            "x-chroma-token": str(tokens.get("CHROMA_API_KEY")),
+        },
+    )
+
+
+def _get_metadata(
+    item: Project,
+    metadata_keys: list[str],
+) -> dict:
+    item_dict: dict = item.dict()
+    return {
+        key: item_dict[key]
+        for key in metadata_keys
+        if key in item_dict and item_dict[key] is not None
+    }
+
+
+def add_item(
+    item: Project,
+    item_id: str,
+    item_document: str,
+    metadata_keys: list[str],
+    collection_name: str,
+    client: chromadb.api.client.Client,
+) -> None:
+    span_name: str = f"{SPAN_KEY}-add_item"
+    with tracer.start_as_current_span(span_name) as span:
+
+        def _setup_medata() -> dict:
+            _metadata = _get_metadata(
+                item=item,
+                metadata_keys=metadata_keys,
+            )
+            _metadata.update(
+                {
+                    "collection_name": collection_name,
+                },
+            )
+            return _metadata
+
+        span.add_event(
+            name=f"{span_name}-queued",
+            attributes={
+                "item_id": item_id,
+                "collection_name": collection_name,
+            },
+        )
+        metadata: dict = _setup_medata()
+        span.set_attributes(
+            attributes=metadata,
+        )
+        span.add_event(
+            name=f"{span_name}-started",
+            attributes=metadata,
+        )
+        collection: chromadb.Collection = client.get_or_create_collection(
+            name=collection_name,
+        )
+        collection.add(
+            ids=[item_id],
+            documents=[item_document],
+            metadatas=[metadata],
+        )
+        span.add_event(
+            name=f"{span_name}-completed",
+            attributes=metadata,
+        )
+
+
+def get_items(
+    query: str,
+    n_results: int,
+    collection_name: str,
+    client: chromadb.api.client.Client,
+) -> chromadb.QueryResult:
+    with tracer.start_as_current_span("get_items") as span:
+        attributes: dict = {
+            "query": query,
+            "collection_name": collection_name,
+            "n_results": n_results,
+        }
+        span.set_attributes(attributes)
+        span.add_event(
+            name="get_items-started",
+            attributes=attributes,
+        )
+        span.add_event(
+            name="get_collection-started",
+            attributes=attributes,
+        )
+        collection: chromadb.Collection = client.get_collection(
+            name=collection_name,
+        )
+        span.add_event(
+            name="get_collection-completed",
+            attributes={
+                "collection_name": collection_name,
+            },
+        )
+        span.add_event(
+            name="query-started",
+            attributes=attributes,
+        )
+        result: chromadb.QueryResult = collection.query(
+            query_texts=query,
+            n_results=n_results,
+        )
+        span.add_event(
+            name="query-completed",
+        )
+        return result

hackathon_project_tracker/hackathon_project_tracker/helper_flask/__init__.py

@@ -0,0 +1,3 @@
+from .helper_flask import ResponseCode, ResponseStatus
+
+__all__ = ["ResponseCode", "ResponseStatus"]

hackathon_project_tracker/hackathon_project_tracker/helper_flask/helper_flask.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from enum import Enum
+
+from pydantic.dataclasses import dataclass
+
+from .pydantic_settings import BaseModel  # trunk-ignore(ruff/TCH001)
+
+
+class ResponseCode(Enum):
+    """Enumeration of response codes used in the application."""
+
+    ACCEPTED_AND_SUCCESSFUL = 200
+    ACCEPTED_WITH_NOTHING_TO_DO = 202
+    FAILURE_WITH_RETRY = 429
+    FAILURE_WITH_NO_RETRY = 405
+
+
+class PydanticConfiguration:
+    """Configuration class for Pydantic model settings.
+
+    This class defines configuration options for Pydantic models used within the application.
+
+    Attributes:
+        arbitrary_types_allowed (bool): When True, allows Pydantic to work with arbitrary Python types
+            that it doesn't natively support. This is useful when working with custom classes
+            or third-party types that need to be included in Pydantic models.
+    """
+    arbitrary_types_allowed = True
+
+
+@dataclass(
+    config=PydanticConfiguration,
+)
+class ResponseStatus:
+    """Represents the status of a response, including the response code, response string, and an optional response dictionary."""
+
+    response_string: str | None
+    response_code: ResponseCode
+    response_dict: BaseModel | None = None
+
+    def __repr__(
+        self: ResponseStatus,
+    ) -> str:
+        """Returns a string representation of the ResponseStatus object."""
+        return f"{self.response_code.value} : {self.response_code.name} : {self.response_string}"
+
+    @classmethod
+    def from_response_code(
+        cls: type[ResponseStatus],
+        response_code: ResponseCode,
+    ) -> ResponseStatus:
+        return ResponseStatus(
+            response_string=response_code.name,
+            response_code=response_code,
+        )
+
+    def get_return_response_dict(
+        self: ResponseStatus,
+    ) -> BaseModel:
+        response_dict: BaseModel | None = self.response_dict
+        if response_dict is None:
+            raise AttributeError
+
+        return response_dict
+
+    def get_return_response_with_dict(
+        self: ResponseStatus,
+    ) -> tuple[BaseModel, int]:
+        return self.get_return_response_dict(), self.response_code.value
+
+    def get_return_response_string(
+        self: ResponseStatus,
+    ) -> str:
+        response_string: str | None = self.response_string
+        if response_string is None:
+            raise AttributeError
+
+        return response_string
+
+    def get_return_response_tuple_with_string(
+        self: ResponseStatus,
+    ) -> tuple[str, int]:
+        return self.get_return_response_string(), self.response_code.value

hackathon_project_tracker/hackathon_project_tracker/helper_flask/pydantic_settings.py

@@ -0,0 +1,16 @@
+from pydantic import (
+    BaseModel as BaseModelPydantic,
+    ConfigDict,
+)
+
+
+class BaseModel(BaseModelPydantic):
+    """A base model for all models in the application.
+    We validate whenever a value is assigned to an attribute.
+    We also allow populating attributes by their name/alias.
+    """
+
+    model_config = ConfigDict(
+        validate_assignment=True,
+        populate_by_name=True,
+    )

hackathon_project_tracker/hackathon_project_tracker/helper_logging/__init__.py

@@ -0,0 +1,21 @@
+__all__ = [
+    "CustomError",
+    "Logger",
+    "Severity",
+    "log",
+    "log_an_exception",
+    "log_with_exception",
+    "setup_logger",
+    "get_otel_config",
+]
+
+from .helper_logging import (
+    CustomError,
+    Logger,
+    Severity,
+    log,
+    log_an_exception,
+    log_with_exception,
+    setup_logger,
+)
+from .helper_otel import get_otel_config