15933171487

Committed 27 Jun 2025 06:19PM UTC coverage: 94.349%. Remained the same

Build # 15933171487

Build Type

push

github

Committed by

web-flow

Commit Message

Fix ruff error (#1565)

Run Details

119 of 125 branches covered (95.2%)

Branch coverage included in aggregate %.

1367 of 1450 relevant lines covered (94.28%)

0.94 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 AsyncBase(AsyncAttrs, DeclarativeBase):
    """Base class for SQLAlchemy models."""


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


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[T: AsyncBase](
    *,
    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 AsyncBase(AsyncAttrs, DeclarativeBase):	1✔
19	"""Base class for SQLAlchemy models."""
20
21
22	T = TypeVar("T", bound=AsyncBase)	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[T: AsyncBase](	1✔
32	*,
33	session: AsyncSession,
34	data_list: list[T],
35	) -> None:
36	"""Merge (upsert) or insert data."""
37	successful_merges = 0	1✔
38	failed_merges = 0	1✔
39
40	try:	1✔
41	for item in data_list:	1✔
42	try:	1✔
43	await session.merge(item) # Merge (upsert)	1✔
44	successful_merges += 1	1✔
45	except ValueError:	1✔
46	# This is likely the NumPy array truth value error
47	failed_merges += 1	×
48	LOGGER.exception(f"Failed to merge item ({type(item).__name__})")	×
49	continue	×
50	except Exception:	1✔
51	failed_merges += 1	1✔
52	LOGGER.exception(	1✔
53	f"Unexpected error merging item ({type(item).__name__})"
54	)
55	continue	1✔
56
57	# Commit once after processing all items
58	if successful_merges > 0:	1✔
59	await session.commit()	1✔
60	LOGGER.debug(	1✔
61	f"Committed {successful_merges} items (skipped {failed_merges})"
62	)
63	else:
64	await session.rollback()	1✔
65	LOGGER.warning(	1✔
66	f"No successful merges to commit, all {failed_merges} items failed"
67	)
68
69	except Exception:	×
70	# If any exception occurs, roll back the transaction
71	await session.rollback()	×
72	LOGGER.exception("Error during upsert operation")	×
73	raise	×

ggravlingen / pypmanager / 15933171487

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