16137404839

Committed 08 Jul 2025 07:58AM UTC coverage: 69.264% (-2.2%) from 71.494%

Build # 16137404839

Build Type

push

github

Committed by

web-flow

Commit Message

More extensive logging (#135)

Run Details

9 of 29 new or added lines in 5 files covered. (31.03%)

6 existing lines in 3 files now uncovered.

320 of 462 relevant lines covered (69.26%)

1.39 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.77

/src/sabogaapi/api_v1/models.py

"""Beanie database models."""

import datetime
from typing import Annotated, Any, List

from beanie import Document, Indexed, TimeSeriesConfig
from pydantic import BaseModel, Field

from sabogaapi.logger import configure_logger

logger = configure_logger()


class RankHistory(Document):
    date: datetime.datetime = Field(default_factory=datetime.datetime.now)
    bgg_id: int
    bgg_rank: int | None = None
    bgg_geek_rating: float | None = None
    bgg_average_rating: float | None = None

    class Settings:
        timeseries = TimeSeriesConfig(
            time_field="date",
            meta_field="bgg_id",
            bucket_rounding_second=86400,
            bucket_max_span_seconds=86400,
        )
        name = "rank_history"


class Category(BaseModel):
    name: str
    bgg_id: int


class Family(BaseModel):
    name: str
    bgg_id: int


class Mechanic(BaseModel):
    name: str
    bgg_id: int


class Designer(BaseModel):
    name: str
    bgg_id: int


class Boardgame(Document):
    bgg_id: Annotated[int, Indexed(unique=True)]
    bgg_rank: Annotated[int, Indexed()]
    name: str = ""
    bgg_geek_rating: float | None = None
    bgg_average_rating: float | None = None
    description: str | None = None
    image_url: str | None = None
    thumbnail_url: str | None = None
    year_published: int | None = None
    minplayers: int | None = None
    maxplayers: int | None = None
    playingtime: int | None = None
    minplaytime: int | None = None
    maxplaytime: int | None = None
    categories: List[Category] = []
    families: List["Family"] = []
    mechanics: List[Mechanic] = []
    designers: List[Designer] = []

    @staticmethod
    async def get_top_ranked_boardgames(
        compare_to: datetime.datetime,
        page: int = 1,
        page_size: int = 50,
    ) -> List[dict[str, Any]]:
        logger.debug(
            "Fetching top ranked boardgames",
            extra={
                "compare_to": compare_to.isoformat(),
                "page": page,
                "page_size": page_size,
            },
        )

        find_rank_comparison = [
            {"$sort": {"bgg_rank": 1}},
            {"$skip": (page - 1) * page_size},
            {"$limit": page_size},
            {
                "$lookup": {
                    "from": f"{RankHistory.Settings.name}",
                    "let": {"bgg_id": "$bgg_id"},
                    "pipeline": [
                        {
                            "$match": {
                                "$expr": {
                                    "$and": [
                                        {"$eq": ["$bgg_id", "$$bgg_id"]},
                                    ]
                                }
                            }
                        },
                        {"$sort": {"date": -1}},
                        {
                            "$match": {
                                "date": {"$lt": compare_to + datetime.timedelta(days=1)}
                            }
                        },
                        {"$limit": 1},
                    ],
                    "as": "rank_history",
                }
            },
            {"$unwind": {"path": "$rank_history", "preserveNullAndEmptyArrays": True}},
            {
                "$set": {
                    "bgg_rank_change": {
                        "$subtract": ["$rank_history.bgg_rank", "$bgg_rank"]
                    },
                    "bgg_average_rating_change": {
                        "$subtract": [
                            "$bgg_average_rating",
                            "$rank_history.bgg_average_rating",
                        ]
                    },
                    "bgg_geek_rating_change": {
                        "$subtract": [
                            "$bgg_geek_rating",
                            "$rank_history.bgg_geek_rating",
                        ]
                    },
                }
            },
        ]

        rank_data = await Boardgame.aggregate(
            aggregation_pipeline=find_rank_comparison
        ).to_list()
        logger.info(
            "Top ranked boardgames fetched",
            extra={
                "returned_count": len(rank_data),
                "page": page,
                "page_size": page_size,
            },
        )

        return rank_data

    @staticmethod
    async def get_boardgame_with_historical_data(
        bgg_id: int,
        start_date: datetime.datetime,
        end_date: datetime.datetime,
        mode: str,
    ) -> dict | None:
        logger.debug(
            "Fetching boardgame with historical data",
            extra={
                "bgg_id": bgg_id,
                "start_date": start_date.isoformat(),
                "end_date": end_date.isoformat(),
                "mode": mode,
            },
        )
        date_diff = (end_date - start_date).days
        if mode == "auto":
            if date_diff <= 30:
                mode = "daily"
            elif date_diff <= 180:
                mode = "weekly"
            else:
                mode = "yearly"
            logger.debug(
                "Auto-detected mode", extra={"mode": mode, "date_diff_days": date_diff}
            )

        pipeline = [
            {"$match": {"bgg_id": bgg_id}},
            {
                "$lookup": {
                    "from": f"{RankHistory.Settings.name}",
                    "let": {"bgg_id": "$bgg_id"},
                    "pipeline": [
                        {
                            "$match": {
                                "$expr": {"$eq": ["$bgg_id", "$$bgg_id"]},
                                "date": {"$lte": end_date, "$gte": start_date},
                            }
                        },
                        {"$sort": {"date": 1}},
                    ],
                    "as": "bgg_rank_history",
                }
            },
        ]

        result = await Boardgame.aggregate(aggregation_pipeline=pipeline).to_list()

        if not result:
            logger.warning("No boardgame found with bgg_id", extra={"bgg_id": bgg_id})
            return None

        boardgame_data = result[0]

        # Apply mode filtering to bgg_rank_history
        history = boardgame_data["bgg_rank_history"]
        if mode == "weekly":
            history = history[::7]  # Every 7th entry
        elif mode == "yearly":
            seen_years = set()
            yearly_history = []
            for entry in history:
                year = entry["date"].year
                if year not in seen_years:
                    yearly_history.append(entry)
                    seen_years.add(year)
            history = yearly_history

        boardgame_data["bgg_rank_history"] = [
            {
                "date": entry["date"].replace(
                    hour=0, minute=0, second=0, microsecond=0
                ),
                "bgg_id": entry["bgg_id"],
                "bgg_rank": entry["bgg_rank"],
                "bgg_geek_rating": entry["bgg_geek_rating"],
                "bgg_average_rating": entry["bgg_average_rating"],
            }
            for entry in history
        ]
        logger.info(
            "Boardgame with historical data returned",
            extra={
                "bgg_id": bgg_id,
                "history_points": len(boardgame_data["bgg_rank_history"]),
                "mode": mode,
            },
        )
        return boardgame_data

    class Settings:
        name = "boardgames"

1	"""Beanie database models."""
2
3	import datetime	2✔
4	from typing import Annotated, Any, List	2✔
5
6	from beanie import Document, Indexed, TimeSeriesConfig	2✔
7	from pydantic import BaseModel, Field	2✔
8
9	from sabogaapi.logger import configure_logger	2✔
10
11	logger = configure_logger()	2✔
12
13
14	class RankHistory(Document):	2✔
15	date: datetime.datetime = Field(default_factory=datetime.datetime.now)	2✔
16	bgg_id: int	2✔
17	bgg_rank: int \| None = None	2✔
18	bgg_geek_rating: float \| None = None	2✔
19	bgg_average_rating: float \| None = None	2✔
20
21	class Settings:	2✔
22	timeseries = TimeSeriesConfig(	2✔
23	time_field="date",
24	meta_field="bgg_id",
25	bucket_rounding_second=86400,
26	bucket_max_span_seconds=86400,
27	)
28	name = "rank_history"	2✔
29
30
31	class Category(BaseModel):	2✔
32	name: str	2✔
33	bgg_id: int	2✔
34
35
36	class Family(BaseModel):	2✔
37	name: str	2✔
38	bgg_id: int	2✔
39
40
41	class Mechanic(BaseModel):	2✔
42	name: str	2✔
43	bgg_id: int	2✔
44
45
46	class Designer(BaseModel):	2✔
47	name: str	2✔
48	bgg_id: int	2✔
49
50
51	class Boardgame(Document):	2✔
52	bgg_id: Annotated[int, Indexed(unique=True)]	2✔
53	bgg_rank: Annotated[int, Indexed()]	2✔
54	name: str = ""	2✔
55	bgg_geek_rating: float \| None = None	2✔
56	bgg_average_rating: float \| None = None	2✔
57	description: str \| None = None	2✔
58	image_url: str \| None = None	2✔
59	thumbnail_url: str \| None = None	2✔
60	year_published: int \| None = None	2✔
61	minplayers: int \| None = None	2✔
62	maxplayers: int \| None = None	2✔
63	playingtime: int \| None = None	2✔
64	minplaytime: int \| None = None	2✔
65	maxplaytime: int \| None = None	2✔
66	categories: List[Category] = []	2✔
67	families: List["Family"] = []	2✔
68	mechanics: List[Mechanic] = []	2✔
69	designers: List[Designer] = []	2✔
70
71	@staticmethod	2✔
72	async def get_top_ranked_boardgames(	2✔
73	compare_to: datetime.datetime,
74	page: int = 1,
75	page_size: int = 50,
76	) -> List[dict[str, Any]]:
NEW 77	logger.debug(	×
78	"Fetching top ranked boardgames",
79	extra={
80	"compare_to": compare_to.isoformat(),
81	"page": page,
82	"page_size": page_size,
83	},
84	)
85
UNCOV 86	find_rank_comparison = [	×
87	{"$sort": {"bgg_rank": 1}},
88	{"$skip": (page - 1) * page_size},
89	{"$limit": page_size},
90	{
91	"$lookup": {
92	"from": f"{RankHistory.Settings.name}",
93	"let": {"bgg_id": "$bgg_id"},
94	"pipeline": [
95	{
96	"$match": {
97	"$expr": {
98	"$and": [
99	{"$eq": ["$bgg_id", "$$bgg_id"]},
100	]
101	}
102	}
103	},
104	{"$sort": {"date": -1}},
105	{
106	"$match": {
107	"date": {"$lt": compare_to + datetime.timedelta(days=1)}
108	}
109	},
110	{"$limit": 1},
111	],
112	"as": "rank_history",
113	}
114	},
115	{"$unwind": {"path": "$rank_history", "preserveNullAndEmptyArrays": True}},
116	{
117	"$set": {
118	"bgg_rank_change": {
119	"$subtract": ["$rank_history.bgg_rank", "$bgg_rank"]
120	},
121	"bgg_average_rating_change": {
122	"$subtract": [
123	"$bgg_average_rating",
124	"$rank_history.bgg_average_rating",
125	]
126	},
127	"bgg_geek_rating_change": {
128	"$subtract": [
129	"$bgg_geek_rating",
130	"$rank_history.bgg_geek_rating",
131	]
132	},
133	}
134	},
135	]
136
137	rank_data = await Boardgame.aggregate(	×
138	aggregation_pipeline=find_rank_comparison
139	).to_list()
NEW 140	logger.info(	×
141	"Top ranked boardgames fetched",
142	extra={
143	"returned_count": len(rank_data),
144	"page": page,
145	"page_size": page_size,
146	},
147	)
148
149	return rank_data	×
150
151	@staticmethod	2✔
152	async def get_boardgame_with_historical_data(	2✔
153	bgg_id: int,
154	start_date: datetime.datetime,
155	end_date: datetime.datetime,
156	mode: str,
157	) -> dict \| None:
NEW 158	logger.debug(	×
159	"Fetching boardgame with historical data",
160	extra={
161	"bgg_id": bgg_id,
162	"start_date": start_date.isoformat(),
163	"end_date": end_date.isoformat(),
164	"mode": mode,
165	},
166	)
167	date_diff = (end_date - start_date).days	×
168	if mode == "auto":	×
169	if date_diff <= 30:	×
170	mode = "daily"	×
171	elif date_diff <= 180:	×
172	mode = "weekly"	×
173	else:
174	mode = "yearly"	×
NEW 175	logger.debug(	×
176	"Auto-detected mode", extra={"mode": mode, "date_diff_days": date_diff}
177	)
178
179	pipeline = [	×
180	{"$match": {"bgg_id": bgg_id}},
181	{
182	"$lookup": {
183	"from": f"{RankHistory.Settings.name}",
184	"let": {"bgg_id": "$bgg_id"},
185	"pipeline": [
186	{
187	"$match": {
188	"$expr": {"$eq": ["$bgg_id", "$$bgg_id"]},
189	"date": {"$lte": end_date, "$gte": start_date},
190	}
191	},
192	{"$sort": {"date": 1}},
193	],
194	"as": "bgg_rank_history",
195	}
196	},
197	]
198
199	result = await Boardgame.aggregate(aggregation_pipeline=pipeline).to_list()	×
200
201	if not result:	×
NEW 202	logger.warning("No boardgame found with bgg_id", extra={"bgg_id": bgg_id})	×
UNCOV 203	return None	×
204
205	boardgame_data = result[0]	×
206
207	# Apply mode filtering to bgg_rank_history
208	history = boardgame_data["bgg_rank_history"]	×
209	if mode == "weekly":	×
210	history = history[::7] # Every 7th entry	×
211	elif mode == "yearly":	×
212	seen_years = set()	×
213	yearly_history = []	×
214	for entry in history:	×
215	year = entry["date"].year	×
216	if year not in seen_years:	×
217	yearly_history.append(entry)	×
218	seen_years.add(year)	×
219	history = yearly_history	×
220
221	boardgame_data["bgg_rank_history"] = [	×
222	{
223	"date": entry["date"].replace(
224	hour=0, minute=0, second=0, microsecond=0
225	),
226	"bgg_id": entry["bgg_id"],
227	"bgg_rank": entry["bgg_rank"],
228	"bgg_geek_rating": entry["bgg_geek_rating"],
229	"bgg_average_rating": entry["bgg_average_rating"],
230	}
231	for entry in history
232	]
NEW 233	logger.info(	×
234	"Boardgame with historical data returned",
235	extra={
236	"bgg_id": bgg_id,
237	"history_points": len(boardgame_data["bgg_rank_history"]),
238	"mode": mode,
239	},
240	)
UNCOV 241	return boardgame_data	×
242
243	class Settings:	2✔
244	name = "boardgames"	2✔

Ruberhauptmann / saboga-api / 16137404839

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous