Add op module for transaction

[jason810496]

No description provided.

[gemini-code-assist]

Summary of Changes

Hello @jason810496, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces a significant architectural enhancement by creating a dedicated op module to encapsulate all direct SQL interactions with PGMQ. This refactoring improves code organization and maintainability by separating the concerns of SQL execution from the PGMQueue class. It also provides greater flexibility for users by allowing them to pass their own SQLAlchemy sessions and explicitly manage transaction commits through the new op module's static methods.

Highlights

• New op Module Introduced: A new module, pgmq_sqlalchemy.operation, has been added and exposed as op. This module centralizes all direct PGMQ SQL operations, providing a clear separation of concerns.
• Refactored PGMQueue Class: The existing PGMQueue class has been refactored to delegate its core PGMQ interactions to the new op module, significantly simplifying its internal implementation and making it leaner.
• Explicit Session Management: Public methods within PGMQueue now accept an optional session parameter, allowing users to explicitly manage SQLAlchemy sessions and control transaction commits for more granular control over database interactions.
• Synchronous and Asynchronous Operations: The new op module provides both synchronous (Session) and asynchronous (AsyncSession) static methods for all PGMQ operations, catering to different application architectures.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request refactors the PGMQ database operations into a new pgmq_sqlalchemy.operation module. This is a great architectural improvement that centralizes database logic and improves separation of concerns. The PGMQueue class is updated to use this new module, and its methods now accept an optional session argument, giving users more control over transactions.

My review focuses on a few critical issues in the new operation module, including a SQL injection vulnerability and a bug that would cause a TypeError at runtime. I've also included a suggestion to reduce code duplication in the queue module. Addressing these points will significantly improve the security and robustness of the library.

[gemini-code-assist]

The methods _get_send_statement and _get_send_batch_statement use f-strings to construct SQL queries. This is a security risk as it can lead to SQL injection vulnerabilities through the queue_name parameter. You should use parameterized queries for all user-provided inputs to prevent this.

This change will also require updating the send, send_async, send_batch, and send_batch_async methods to handle parameters instead of a raw SQL string.

    def _get_send_statement(
        queue_name: str, message: dict, delay: int
    ) -> Tuple[str, Dict[str, Any]]:
        """Get statement and params for send."""
        return (
            "select * from pgmq.send(:queue_name, :message::jsonb, :delay);",
            {
                "queue_name": queue_name,
                "message": __import__("json").dumps(message),
                "delay": delay,
            },
        )

    @staticmethod
    def _get_send_batch_statement(
        queue_name: str, messages: List[dict], delay: int
    ) -> Tuple[str, Dict[str, Any]]:
        """Get statement and params for send_batch."""
        return (
            "select * from pgmq.send_batch(:queue_name, :messages, :delay);",
            {
                "queue_name": queue_name,
                "messages": [__import__("json").dumps(m) for m in messages],
                "delay": delay,
            },
        )

[gemini-code-assist]

The QueueMetrics data class is being initialized with a scrape_time argument, but the class definition in pgmq_sqlalchemy/schema.py does not include this field. This will cause a TypeError at runtime. This issue is also present in metrics_async, metrics_all, and metrics_all_async methods. You should remove the scrape_time argument from the QueueMetrics constructor in all these methods.

        return QueueMetrics(
            queue_name=row[0],
            queue_length=row[1],
            newest_msg_age_sec=row[2],
            oldest_msg_age_sec=row[3],
            total_messages=row[4],
        )

[gemini-code-assist]

There is a significant amount of boilerplate code for handling synchronous and asynchronous execution, as well as session management, duplicated across almost all methods in this class. This makes the code harder to read and maintain. Consider refactoring this logic into a private helper method or a decorator to reduce duplication.

For example, you could have a helper method like this:

def _execute(self, op_sync, op_async, session, commit, *args, **kwargs):
    if self.is_async:
        if session is None:
            async def _run():
                async with self.session_maker() as s:
                    return await op_async(*args, **kwargs, session=s, commit=commit)
            return self.loop.run_until_complete(_run())
        return self.loop.run_until_complete(op_async(*args, **kwargs, session=session, commit=commit))
    
    if session is None:
        with self.session_maker() as s:
            return op_sync(*args, **kwargs, session=s, commit=commit)
    return op_sync(*args, **kwargs, session=session, commit=commit)

And then create_queue would become:

def create_queue(self, queue_name: str, unlogged: bool = False, *, session: Optional[SESSION_TYPE] = None, commit: bool = True) -> None:
    self._execute(
        PGMQOperation.create_queue,
        PGMQOperation.create_queue_async,
        session,
        commit,
        queue_name,
        unlogged=unlogged,
    )

[codecov]

Codecov Report

❌ Patch coverage is 98.14815% with 1 line in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
pgmq_sqlalchemy/queue.py	98.14%	1 Missing ⚠️

Files with missing lines	Coverage Δ
pgmq_sqlalchemy/queue.py	98.26% <98.14%> (+0.39%)	⬆️

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[jason810496]

@copilot

1. resolve all the code review comment
2. finish tests/test_operation.py

[Copilot]

@jason810496 I've opened a new pull request, #22, to work on those changes. Once the pull request is ready, I'll request review from you.

[jason810496]

@copilot improve the test coverage:

https://app.codecov.io/gh/jason810496/pgmq-sqlalchemy/pull/21/blob/pgmq_sqlalchemy/queue.py

[Copilot]

@jason810496 I've opened a new pull request, #23, to work on those changes. Once the pull request is ready, I'll request review from you.

[jason810496]

@copilot

Current CI doesn't reflect the failure, I check the logs and it turn out that we got a lot of failure.
Here is one of the failures. However, all the Integration test mark as green instead of red. So we need to fix the CI config IMO.

=================================== FAILURES ===================================
_______________________ test_send_and_read_sync[pg8000] ________________________

self = <sqlalchemy.engine.base.Connection object at 0x7f645f035cd0>
dialect = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x7f645f15a4c0>
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x7f645f035e80>
statement = <sqlalchemy.dialects.postgresql.base.PGCompiler object at 0x7f645f035d30>
parameters = [{'delay': 0, 'message': {'foo': 'bar', 'hello': 'world'}, 'queue_name': 'test_queue_4ed68c3a9f6243f59745e5f6a79d7b53'}]

    def _exec_single_context(
        self,
        dialect: Dialect,
        context: ExecutionContext,
        statement: Union[str, Compiled],
        parameters: Optional[_AnyMultiExecuteParams],
    ) -> CursorResult[Any]:
        """continue the _execute_context() method for a single DBAPI
        cursor.execute() or cursor.executemany() call.
    
        """
        if dialect.bind_typing is BindTyping.SETINPUTSIZES:
            generic_setinputsizes = context._prepare_set_input_sizes()
    
            if generic_setinputsizes:
                try:
                    dialect.do_set_input_sizes(
                        context.cursor, generic_setinputsizes, context
                    )
                except BaseException as e:
                    self._handle_dbapi_exception(
                        e, str(statement), parameters, None, context
                    )
    
        cursor, str_statement, parameters = (
            context.cursor,
            context.statement,
            context.parameters,
        )
    
        effective_parameters: Optional[_AnyExecuteParams]
    
        if not context.executemany:
            effective_parameters = parameters[0]
        else:
            effective_parameters = parameters
    
        if self._has_events or self.engine._has_events:
            for fn in self.dispatch.before_cursor_execute:
                str_statement, effective_parameters = fn(
                    self,
                    cursor,
                    str_statement,
                    effective_parameters,
                    context,
                    context.executemany,
                )
    
        if self._echo:
            self._log_info(str_statement)
    
            stats = context._get_cache_stats()
    
            if not self.engine.hide_parameters:
                self._log_info(
                    "[%s] %r",
                    stats,
                    sql_util._repr_params(
                        effective_parameters,
                        batches=10,
                        ismulti=context.executemany,
                    ),
                )
            else:
                self._log_info(
                    "[%s] [SQL parameters hidden due to hide_parameters=True]",
                    stats,
                )
    
        evt_handled: bool = False
        try:
            if context.execute_style is ExecuteStyle.EXECUTEMANY:
                effective_parameters = cast(
                    "_CoreMultiExecuteParams", effective_parameters
                )
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_executemany:
                        if fn(
                            cursor,
                            str_statement,
                            effective_parameters,
                            context,
                        ):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_executemany(
                        cursor,
                        str_statement,
                        effective_parameters,
                        context,
                    )
            elif not effective_parameters and context.no_parameters:
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute_no_params:
                        if fn(cursor, str_statement, context):
                            evt_handled = True
                            break
                if not evt_handled:
                    self.dialect.do_execute_no_params(
                        cursor, str_statement, context
                    )
            else:
                effective_parameters = cast(
                    "_CoreSingleExecuteParams", effective_parameters
                )
                if self.dialect._has_events:
                    for fn in self.dialect.dispatch.do_execute:
                        if fn(
                            cursor,
                            str_statement,
                            effective_parameters,
                            context,
                        ):
                            evt_handled = True
                            break
                if not evt_handled:
>                   self.dialect.do_execute(
                        cursor, str_statement, effective_parameters, context
                    )

.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1967: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x7f645f15a4c0>
cursor = <cursor object at 0x7f645f2c4e50; closed: -1>
statement = 'select * from pgmq.send(%(queue_name)s, CAST(%(message)s AS jsonb), %(delay)s);'
parameters = {'delay': 0, 'message': {'foo': 'bar', 'hello': 'world'}, 'queue_name': 'test_queue_4ed68c3a9f6243f59745e5f6a79d7b53'}
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x7f645f035e80>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       psycopg2.ProgrammingError: can't adapt type 'dict'

.venv/lib/python3.9/site-packages/sqlalchemy/engine/default.py:952: ProgrammingError

The above exception was the direct cause of the following exception:

get_session_maker = sessionmaker(class_='Session', bind=Engine(***localhost:5432/pgmq_py39_psycopg2), autoflush=True, expire_on_commit=True)
db_session = <sqlalchemy.orm.session.Session object at 0x7f645f0356a0>

    def test_send_and_read_sync(get_session_maker, db_session):
        """Test sending and reading messages."""
        queue_name = f"test_queue_{uuid.uuid4().hex}"
    
        # Create queue
        with get_session_maker() as session:
            PGMQOperation.create_queue(queue_name, unlogged=False, session=session, commit=True)
    
        # Send a message
        with get_session_maker() as session:
>           msg_id = PGMQOperation.send(queue_name, MSG, delay=0, session=session, commit=True)

tests/test_operation.py:100: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
pgmq_sqlalchemy/operation.py:587: in send
    row = session.execute(text(stmt), params).fetchone()
.venv/lib/python3.9/site-packages/sqlalchemy/orm/session.py:2351: in execute
    return self._execute_internal(
.venv/lib/python3.9/site-packages/sqlalchemy/orm/session.py:2258: in _execute_internal
    result = conn.execute(
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1419: in execute
    return meth(
.venv/lib/python3.9/site-packages/sqlalchemy/sql/elements.py:527: in _execute_on_connection
    return connection._execute_clauseelement(
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1641: in _execute_clauseelement
    ret = self._execute_context(
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1846: in _execute_context
    return self._exec_single_context(
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1986: in _exec_single_context
    self._handle_dbapi_exception(
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:2363: in _handle_dbapi_exception
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
.venv/lib/python3.9/site-packages/sqlalchemy/engine/base.py:1967: in _exec_single_context
    self.dialect.do_execute(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.dialects.postgresql.psycopg2.PGDialect_psycopg2 object at 0x7f645f15a4c0>
cursor = <cursor object at 0x7f645f2c4e50; closed: -1>
statement = 'select * from pgmq.send(%(queue_name)s, CAST(%(message)s AS jsonb), %(delay)s);'
parameters = {'delay': 0, 'message': {'foo': 'bar', 'hello': 'world'}, 'queue_name': 'test_queue_4ed68c3a9f6243f59745e5f6a79d7b53'}
context = <sqlalchemy.dialects.postgresql.psycopg2.PGExecutionContext_psycopg2 object at 0x7f645f035e80>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       sqlalchemy.exc.ProgrammingError: (psycopg2.ProgrammingError) can't adapt type 'dict'
E       [SQL: select * from pgmq.send(%(queue_name)s, CAST(%(message)s AS jsonb), %(delay)s);]
E       [parameters: {'queue_name': 'test_queue_4ed68c3a9f6243f59745e5f6a79d7b53', 'message': {'foo': 'bar', 'hello': 'world'}, 'delay': 0}]
E       (Background on this error at: https://sqlalche.me/e/20/f405)

.venv/lib/python3.9/site-packages/sqlalchemy/engine/default.py:952: ProgrammingError

[Copilot]

@jason810496 I've opened a new pull request, #24, to work on those changes. Once the pull request is ready, I'll request review from you.

[jason810496]

@copilot Fix the jsonb handling with bindparams, here is the example usage of bindparams.

from sqlalchemy import create_engine, text, bindparam
from sqlalchemy.dialects.postgresql import JSONB
import json

# Replace with your actual database URL
DATABASE_URL = "postgresql://user:password@localhost/mydatabase"
engine = create_engine(DATABASE_URL)

# The Python dictionary you want to insert or pass as a parameter
json_data = {"key": "value", "nested": [1, 2, 3]}

# Define the parameters with explicit type information for the JSONB field
params = {
    'p1': 1,
    'p2': json_data
}

# Construct the text query and bind the parameters
# Example: Calling a function that accepts an integer and a jsonb
# (adjust the SQL to your specific use case, e.g., INSERT or UPDATE)
sql_query = text("SELECT my_func(:p1, :p2)").bindparams(
    bindparam('p2', type_=JSONB)
)

# Execute the query
with engine.connect() as connection:
    result = connection.execute(sql_query, params)
    # Process results if necessary
    # print(result.scalar())

Since we encounter the following error for message: dict and messages: List[dict]:

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       psycopg2.ProgrammingError: can't adapt type 'dict'

.venv/lib/python3.9/site-packages/sqlalchemy/engine/default.py:952: ProgrammingError

The above exception was the direct cause of the following exception:

get_session_maker = sessionmaker(class_='Session', bind=Engine(***localhost:5432/pgmq_py39_psycopg2), autoflush=True, expire_on_commit=True)
db_session = <sqlalchemy.orm.session.Session object at 0x7fe9b231ffa0>

    def test_send_and_read_sync(get_session_maker, db_session):
        """Test sending and reading messages."""
        queue_name = f"test_queue_{uuid.uuid4().hex}"
    
        # Create queue
        with get_session_maker() as session:
            PGMQOperation.create_queue(queue_name, unlogged=False, session=session, commit=True)
    
        # Send a message
        with get_session_maker() as session:
>           msg_id = PGMQOperation.send(queue_name, MSG, delay=0, session=session, commit=True)

tests/test_operation.py:100: 

[Copilot]

@jason810496 I've opened a new pull request, #25, to work on those changes. Once the pull request is ready, I'll request review from you.

[jason810496]

@copilot Fix test error:

FAILED tests/test_queue.py::test_read_archive_empty[pgmq_by_async_dsn] - AttributeError: 'PGMQueue' object has no attribute 'read_archive'
FAILED tests/test_queue.py::test_read_archive_empty[pgmq_by_async_engine] - AttributeError: 'PGMQueue' object has no attribute 'read_archive'
FAILED tests/test_queue.py::test_read_archive_empty[pgmq_by_async_session_maker] - AttributeError: 'PGMQueue' object has no attribute 'read_archive'
FAILED tests/test_queue.py::test_read_archive_batch[pgmq_by_async_dsn] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch[pgmq_by_async_engine] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch[pgmq_by_async_session_maker] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch_empty[pgmq_by_async_dsn] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch_empty[pgmq_by_async_engine] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch_empty[pgmq_by_async_session_maker] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch_limit[pgmq_by_async_dsn] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch_limit[pgmq_by_async_engine] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_read_archive_batch_limit[pgmq_by_async_session_maker] - AttributeError: 'PGMQueue' object has no attribute 'read_archive_batch'
FAILED tests/test_queue.py::test_create_partitioned_queue_invalid_numeric_interval[pgmq_by_async_dsn] - assert 'Numeric partition interval must be positive' in "Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year"
 +  where "Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year" = str(ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year"))
 +    where ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year") = <ExceptionInfo ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year") tblen=7>.value
FAILED tests/test_queue.py::test_create_partitioned_queue_invalid_numeric_interval[pgmq_by_async_engine] - assert 'Numeric partition interval must be positive' in "Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year"
 +  where "Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year" = str(ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year"))
 +    where ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year") = <ExceptionInfo ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year") tblen=7>.value
FAILED tests/test_queue.py::test_create_partitioned_queue_invalid_numeric_interval[pgmq_by_async_session_maker] - assert 'Numeric partition interval must be positive' in "Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year"
 +  where "Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year" = str(ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year"))
 +    where ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year") = <ExceptionInfo ValueError("Invalid time-based partition interval: '-100'. Expected format: '<number> <unit>' where unit is one of: microsecond, millisecond, second, minute, hour, day, week, month, year") tblen=7>.value
====== 35 failed, 136 passed, 11 skipped, 22 warnings in 92.15s (0:01:32) ======

[Copilot]

@jason810496 I've opened a new pull request, #26, to work on those changes. Once the pull request is ready, I'll request review from you.

[jason810496]

@copilot

resolve ambiguous error for pgmq.delete and pgmq.archive:

---

Example error:

FAILED tests/test_queue.py::test_delete_msg[pgmq_by_async_dsn] - sqlalchemy.exc.ProgrammingError: (sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.AmbiguousFunctionError'>: function pgmq.delete(unknown, unknown) is not unique
HINT:  Could not choose a best candidate function. You might need to add explicit type casts.
[SQL: select pgmq.delete($1, $2);]
[parameters: ('test_queue_93a45930e58240e5a4c1d11297364acd', 2)]
(Background on this error at: https://sqlalche.me/e/20/f405)

FAILED tests/test_queue.py::test_detach_archive[pgmq_by_async_session_maker] - sqlalchemy.exc.ProgrammingError: (sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.AmbiguousFunctionError'>: function pgmq.archive(unknown, unknown) is not unique
HINT:  Could not choose a best candidate function. You might need to add explicit type casts.
[SQL: select pgmq.archive($1, $2);]
[parameters: ('test_queue_bc0435ad271d473ba918970c70514019', 1)]

---

Spec

• delete: https://github.com/pgmq/pgmq/blob/main/docs/api/sql/functions.md?plain=1#L372-L437
• archive: https://github.com/pgmq/pgmq/blob/main/docs/api/sql/functions.md?plain=1#L466-L534

[Copilot]

@jason810496 I've opened a new pull request, #27, to work on those changes. Once the pull request is ready, I'll request review from you.

[jason810496]

@copilot

Fix fixture 'pgmq_all_variants' not found error in test_queue

[Copilot]

@jason810496 I've opened a new pull request, #28, to work on those changes. Once the pull request is ready, I'll request review from you.

[jason810496]

@copilot improve test coverage for

1. The following part for _execute_operation

        def _execute_operation(
# ...
             return self.loop.run_until_complete(
                 op_async(*args, session=session, commit=commit, **kwargs)
             )
# ....
        return op_sync(*args, session=session, commit=commit, **kwargs)

2. The case when vt is not given for read_with_poll

[Copilot]

@jason810496 I've opened a new pull request, #29, to work on those changes. Once the pull request is ready, I'll request review from you.