10417671105

Committed 16 Aug 2024 09:19AM UTC coverage: 94.335% (+0.7%) from 93.643%

Build # 10417671105

Build Type

push

github

Committed by

vladimirgorealionstep

Commit Message

fix requirements.txt path in .githubactions

Run Details

383 of 406 relevant lines covered (94.33%)

0.94 hits per line

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

79.69

/data_persistence_repository/sql_repository_async.py

from typing import List, Iterable, Optional
import contextlib
import asyncio

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    AsyncSession,
    AsyncEngine,
    async_sessionmaker,
)
from sqlalchemy import MetaData, select, delete, update
from sqlalchemy.orm import registry
from sqlalchemy.exc import NoResultFound

from .repository_interface import Repository

import logging

logger = logging.getLogger("sql_repository")


class AsyncSqlRepository(Repository):
    metadata_obj = MetaData()
    registry = registry()

    def __init__(self, url: Optional[str] = None, engine: Optional[AsyncEngine] = None):
        """
        Asynchronous SQL repository.
        :param url: SQL URL for the database connection.
        :param engine: Async engine, if already created.
        """
        if engine is None and url is None:
            raise ValueError("Either url or engine must be provided")

        self._engine = engine if engine else create_async_engine(url, echo=True, pool_recycle=3600, pool_pre_ping=True)
        self._session_factory = async_sessionmaker(self._engine, expire_on_commit=False, class_=AsyncSession)

    async def get_session(self):
        return self._session_factory()

    @contextlib.asynccontextmanager
    async def start_session(self, rollback=True):
        """
            use this context manager mainly
            retries in case of bad connection
        """
        async with self._session_factory() as session:
            try:
                async with session.begin():
                    yield session
            except Exception as e:
                if rollback:
                    await session.rollback()
                logger.error(f"Exception during session: {str(e)}")
                raise

    async def sync_schema(self):
        """Asynchronously create tables and run migrations."""
        async with self._engine.begin() as conn:
            await conn.run_sync(self.metadata_obj.create_all)

    async def add(self, session: AsyncSession, instance: object):
        """Asynchronously save an object."""
        session.add(instance)

    async def add_bulk(self, session: AsyncSession, objects: List[object]):
        """Asynchronously save a list of objects."""
        session.add_all(objects)

    async def get(self, session: AsyncSession, model, **kwargs):
        """Asynchronously get an object."""
        try:
            result = await session.execute(select(model).filter_by(**kwargs))
            return result.scalar_one()
        except NoResultFound:
            return None

    async def exists(self, session: AsyncSession, model, **kwargs) -> bool:
        """Asynchronously check if an object exists."""
        result = await session.execute(select(model).filter_by(**kwargs))
        return result.scalar_one_or_none() is not None

    async def delete(self, session: AsyncSession, model, **kwargs):
        """Asynchronously delete an object."""
        await session.execute(delete(model).filter_by(**kwargs))

    async def filter(self, session: AsyncSession, model, *args, **kwargs) -> Iterable:
        """Asynchronously get a list of objects after applying some filtering."""
        if args and kwargs:
            raise ValueError('Cannot use filter method with both args and kwargs')

        if args:
            result = await session.execute(select(model).filter(*args))
        elif kwargs:
            result = await session.execute(select(model).filter_by(**kwargs))
        else:
            result = await session.execute(select(model))

        # because we're usually using lazy="joined"
        return result.unique().scalars().all()

    async def filter_by_list(self, session: AsyncSession, model, field: str, items_list: List) -> Iterable:
        """Asynchronously filter objects by a list of values in a field."""
        query_field = getattr(model, field)
        result = await session.execute(select(model).filter(query_field.in_(items_list)))
        return result.scalars().all()

    async def patch(self, session: AsyncSession, model, update_data: dict, **kwargs):
        """Asynchronously update specific fields of an object."""
        await session.execute(update(model).filter_by(**kwargs).values(update_data))

1	from typing import List, Iterable, Optional	1✔
2	import contextlib	1✔
3	import asyncio	1✔
4
5	from sqlalchemy.ext.asyncio import (	1✔
6	create_async_engine,
7	AsyncSession,
8	AsyncEngine,
9	async_sessionmaker,
10	)
11	from sqlalchemy import MetaData, select, delete, update	1✔
12	from sqlalchemy.orm import registry	1✔
13	from sqlalchemy.exc import NoResultFound	1✔
14
15	from .repository_interface import Repository	1✔
16
17	import logging	1✔
18
19	logger = logging.getLogger("sql_repository")	1✔
20
21
22	class AsyncSqlRepository(Repository):	1✔
23	metadata_obj = MetaData()	1✔
24	registry = registry()	1✔
25
26	def __init__(self, url: Optional[str] = None, engine: Optional[AsyncEngine] = None):	1✔
27	"""
28	Asynchronous SQL repository.
29	:param url: SQL URL for the database connection.
30	:param engine: Async engine, if already created.
31	"""
32	if engine is None and url is None:	1✔
33	raise ValueError("Either url or engine must be provided")	×
34
35	self._engine = engine if engine else create_async_engine(url, echo=True, pool_recycle=3600, pool_pre_ping=True)	1✔
36	self._session_factory = async_sessionmaker(self._engine, expire_on_commit=False, class_=AsyncSession)	1✔
37
38	async def get_session(self):	1✔
39	return self._session_factory()	×
40
41	@contextlib.asynccontextmanager	1✔
42	async def start_session(self, rollback=True):	1✔
43	"""
44	use this context manager mainly
45	retries in case of bad connection
46	"""
47	async with self._session_factory() as session:	1✔
48	try:	1✔
49	async with session.begin():	1✔
50	yield session	1✔
51	except Exception as e:	1✔
52	if rollback:	1✔
53	await session.rollback()	1✔
54	logger.error(f"Exception during session: {str(e)}")	1✔
55	raise	1✔
56
57	async def sync_schema(self):	1✔
58	"""Asynchronously create tables and run migrations."""
59	async with self._engine.begin() as conn:	×
60	await conn.run_sync(self.metadata_obj.create_all)	×
61
62	async def add(self, session: AsyncSession, instance: object):	1✔
63	"""Asynchronously save an object."""
64	session.add(instance)	1✔
65
66	async def add_bulk(self, session: AsyncSession, objects: List[object]):	1✔
67	"""Asynchronously save a list of objects."""
68	session.add_all(objects)	1✔
69
70	async def get(self, session: AsyncSession, model, **kwargs):	1✔
71	"""Asynchronously get an object."""
72	try:	1✔
73	result = await session.execute(select(model).filter_by(**kwargs))	1✔
74	return result.scalar_one()	1✔
75	except NoResultFound:	×
76	return None	×
77
78	async def exists(self, session: AsyncSession, model, **kwargs) -> bool:	1✔
79	"""Asynchronously check if an object exists."""
80	result = await session.execute(select(model).filter_by(**kwargs))	1✔
81	return result.scalar_one_or_none() is not None	1✔
82
83	async def delete(self, session: AsyncSession, model, **kwargs):	1✔
84	"""Asynchronously delete an object."""
85	await session.execute(delete(model).filter_by(**kwargs))	1✔
86
87	async def filter(self, session: AsyncSession, model, args, *kwargs) -> Iterable:	1✔
88	"""Asynchronously get a list of objects after applying some filtering."""
89	if args and kwargs:	1✔
90	raise ValueError('Cannot use filter method with both args and kwargs')	×
91
92	if args:	1✔
93	result = await session.execute(select(model).filter(*args))	1✔
94	elif kwargs:	×
95	result = await session.execute(select(model).filter_by(**kwargs))	×
96	else:
97	result = await session.execute(select(model))	×
98
99	# because we're usually using lazy="joined"
100	return result.unique().scalars().all()	1✔
101
102	async def filter_by_list(self, session: AsyncSession, model, field: str, items_list: List) -> Iterable:	1✔
103	"""Asynchronously filter objects by a list of values in a field."""
104	query_field = getattr(model, field)	×
105	result = await session.execute(select(model).filter(query_field.in_(items_list)))	×
106	return result.scalars().all()	×
107
108	async def patch(self, session: AsyncSession, model, update_data: dict, **kwargs):	1✔
109	"""Asynchronously update specific fields of an object."""
110	await session.execute(update(model).filter_by(**kwargs).values(update_data))	1✔

smileservices / data_persistence_repository / 10417671105

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