14410217045

Committed 11 Apr 2025 07:02PM UTC coverage: 94.874% (-0.6%) from 95.451%

Build # 14410217045

Build Type

push

github

Committed by

web-flow

Commit Message

Build sync function (#1415)

Run Details

108 of 113 branches covered (95.58%)

Branch coverage included in aggregate %.

1280 of 1350 relevant lines covered (94.81%)

0.95 hits per line

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

82.05

pypmanager/database/utils.py

"""Database utils."""

from __future__ import annotations

import logging
from typing import TypeVar

from sqlalchemy import Connection, inspect
from sqlalchemy.ext.asyncio import (
    AsyncAttrs,
    AsyncSession,
)
from sqlalchemy.orm import DeclarativeBase

LOGGER = logging.getLogger(__name__)


class Base(AsyncAttrs, DeclarativeBase):
    """Base class for SQLAlchemy models."""


T = TypeVar("T", bound=Base)


def check_table_exists(connection: Connection, table_name: str) -> bool:
    """Check if a table exists in the database."""
    inspector = inspect(connection)
    return table_name in inspector.get_table_names()


async def async_upsert_data(*, session: AsyncSession, data_list: list[T]) -> None:
    """Merge (upsert) or insert data."""
    successful_merges = 0
    failed_merges = 0

    try:
        for item in data_list:
            try:
                await session.merge(item)  # Merge (upsert)
                successful_merges += 1
            except ValueError:
                # This is likely the NumPy array truth value error
                failed_merges += 1
                LOGGER.exception(f"Failed to merge item ({type(item).__name__})")
                continue
            except Exception:
                failed_merges += 1
                LOGGER.exception(
                    f"Unexpected error merging item ({type(item).__name__})"
                )
                continue

        # Commit once after processing all items
        if successful_merges > 0:
            await session.commit()
            LOGGER.debug(
                f"Committed {successful_merges} items (skipped {failed_merges})"
            )
        else:
            await session.rollback()
            LOGGER.warning(
                f"No successful merges to commit, all {failed_merges} items failed"
            )

    except Exception:
        # If any exception occurs, roll back the transaction
        await session.rollback()
        LOGGER.exception("Error during upsert operation")
        raise

1	"""Database utils."""
2
3	from __future__ import annotations	1✔
4
5	import logging	1✔
6	from typing import TypeVar	1✔
7
8	from sqlalchemy import Connection, inspect	1✔
9	from sqlalchemy.ext.asyncio import (	1✔
10	AsyncAttrs,
11	AsyncSession,
12	)
13	from sqlalchemy.orm import DeclarativeBase	1✔
14
15	LOGGER = logging.getLogger(__name__)	1✔
16
17
18	class Base(AsyncAttrs, DeclarativeBase):	1✔
19	"""Base class for SQLAlchemy models."""
20
21
22	T = TypeVar("T", bound=Base)	1✔
23
24
25	def check_table_exists(connection: Connection, table_name: str) -> bool:	1✔
26	"""Check if a table exists in the database."""
27	inspector = inspect(connection)	1✔
28	return table_name in inspector.get_table_names()	1✔
29
30
31	async def async_upsert_data(*, session: AsyncSession, data_list: list[T]) -> None:	1✔
32	"""Merge (upsert) or insert data."""
33	successful_merges = 0	1✔
34	failed_merges = 0	1✔
35
36	try:	1✔
37	for item in data_list:	1✔
38	try:	1✔
39	await session.merge(item) # Merge (upsert)	1✔
40	successful_merges += 1	1✔
41	except ValueError:	1✔
42	# This is likely the NumPy array truth value error
43	failed_merges += 1	×
44	LOGGER.exception(f"Failed to merge item ({type(item).__name__})")	×
45	continue	×
46	except Exception:	1✔
47	failed_merges += 1	1✔
48	LOGGER.exception(	1✔
49	f"Unexpected error merging item ({type(item).__name__})"
50	)
51	continue	1✔
52
53	# Commit once after processing all items
54	if successful_merges > 0:	1✔
55	await session.commit()	1✔
56	LOGGER.debug(	1✔
57	f"Committed {successful_merges} items (skipped {failed_merges})"
58	)
59	else:
60	await session.rollback()	1✔
61	LOGGER.warning(	1✔
62	f"No successful merges to commit, all {failed_merges} items failed"
63	)
64
65	except Exception:	×
66	# If any exception occurs, roll back the transaction
67	await session.rollback()	×
68	LOGGER.exception("Error during upsert operation")	×
69	raise	×

ggravlingen / pypmanager / 14410217045

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