Skip to content

Bump PSQLPy and taskiq #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Aug 6, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/test.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ jobs:
pytest:
strategy:
matrix:
py_version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
py_version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
runs-on: "ubuntu-latest"
steps:
- uses: actions/checkout@v4
Expand Down
18 changes: 9 additions & 9 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -12,27 +12,27 @@ repos:
- repo: https://github.com/asottile/add-trailing-comma
rev: v2.1.0
hooks:
- id: add-trailing-comma
- id: add-trailing-comma

- repo: local
hooks:
- id: black
name: Format with Black
entry: poetry run black
language: system
types: [python]

- id: ruff
name: Run ruff lints
entry: poetry run ruff
entry: poetry run ruff check
language: system
pass_filenames: false
types: [python]
types: [ python ]
args:
- "--fix"
- "taskiq_psqlpy"
- "tests"

- id: black
name: Format with Black
entry: poetry run black
language: system
types: [ python ]

- id: mypy
name: Validate types with MyPy
entry: poetry run mypy
Expand Down
1,189 changes: 672 additions & 517 deletions poetry.lock

Large diffs are not rendered by default.

76 changes: 38 additions & 38 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "taskiq-psqlpy"
version = "0.1.5"
version = "0.1.6"
description = "PSQLPy and PostgreSQL integration for taskiq"
authors = ["taskiq-team <taskiq@norely.com>"]
readme = "README.md"
Expand All @@ -9,11 +9,11 @@ classifiers = [
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
]
homepage = "https://github.com/taskiq-python/taskiq-psqlpy"
repository = "https://github.com/taskiq-python/taskiq-psqlpy"
Expand All @@ -28,14 +28,14 @@ keywords = [
]

[tool.poetry.dependencies]
python = "^3.8.1"
psqlpy = "^0.8.7"
taskiq = "^0.11.10"
python = "^3.9.0"
psqlpy = "^0.11.1"
taskiq = "^0.11.17"

[tool.poetry.group.dev.dependencies]
black = "^23.1.0"
pre-commit = "^2.20.0"
mypy = "^1.1.1"
mypy = "^1.16"
flake8 = "^6"
autoflake = "^1.4"
yesqa = "^1.4.0"
Expand All @@ -44,7 +44,7 @@ pytest-xdist = "^3.2.1"
anyio = "^3.6.2"
pytest-cov = "^4.0.0"
wemake-python-styleguide = "^0.18.0"
ruff = "^0.3.4"
ruff = "^0.12"


[tool.mypy]
Expand All @@ -63,68 +63,68 @@ requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

[tool.ruff]
line-length = 88

[tool.ruff.lint]
# List of enabled rulsets.
# See https://docs.astral.sh/ruff/rules/ for more information.
select = [
"E", # Error
"F", # Pyflakes
"W", # Pycodestyle
"E", # Error
"F", # Pyflakes
"W", # Pycodestyle
"C90", # McCabe complexity
"I", # Isort
"N", # pep8-naming
"D", # Pydocstyle
"I", # Isort
"N", # pep8-naming
"D", # Pydocstyle
"ANN", # Pytype annotations
"S", # Bandit
"B", # Bugbear
"S", # Bandit
"B", # Bugbear
"COM", # Commas
"C4", # Comprehensions
"C4", # Comprehensions
"ISC", # Implicit string concat
"PIE", # Unnecessary code
"T20", # Catch prints
"PYI", # validate pyi files
"Q", # Checks for quotes
"Q", # Checks for quotes
"RSE", # Checks raise statements
"RET", # Checks return statements
"SLF", # Self checks
"SIM", # Simplificator
"PTH", # Pathlib checks
"ERA", # Checks for commented out code
"PL", # PyLint checks
"PL", # PyLint checks
"RUF", # Specific to Ruff checks
]
ignore = [
"D105", # Missing docstring in magic method
"D107", # Missing docstring in __init__
"D212", # Multi-line docstring summary should start at the first line
"D401", # First line should be in imperative mood
"D104", # Missing docstring in public package
"D100", # Missing docstring in public module
"ANN102", # Missing type annotation for self in method
"ANN101", # Missing type annotation for argument
"ANN401", # typing.Any are disallowed in `**kwargs
"D105", # Missing docstring in magic method
"D107", # Missing docstring in __init__
"D212", # Multi-line docstring summary should start at the first line
"D401", # First line should be in imperative mood
"D104", # Missing docstring in public package
"D100", # Missing docstring in public module
"ANN401", # typing.Any are disallowed in `**kwargs
"PLR0913", # Too many arguments for function call
"D106", # Missing docstring in public nested class
"D106", # Missing docstring in public nested class
]
exclude = [".venv/"]
mccabe = { max-complexity = 10 }
line-length = 88

[tool.ruff.per-file-ignores]
[tool.ruff.lint.per-file-ignores]
"tests/*" = [
"S101", # Use of assert detected
"S301", # Use of pickle detected
"D103", # Missing docstring in public function
"S101", # Use of assert detected
"S301", # Use of pickle detected
"D103", # Missing docstring in public function
"SLF001", # Private member accessed
"S311", # Standard pseudo-random generators are not suitable for security/cryptographic purposes
"D101", # Missing docstring in public class
"S311", # Standard pseudo-random generators are not suitable for security/cryptographic purposes
"D101", # Missing docstring in public class
]

[tool.ruff.pydocstyle]
[tool.ruff.lint.pydocstyle]
convention = "pep257"
ignore-decorators = ["typing.overload"]

[tool.ruff.pylint]
[tool.ruff.lint.pylint]
allow-magic-value-types = ["int", "str", "float"]

[tool.ruff.flake8-bugbear]
[tool.ruff.lint.flake8-bugbear]
extend-immutable-calls = ["taskiq_dependencies.Depends", "taskiq.TaskiqDepends"]
77 changes: 41 additions & 36 deletions taskiq_psqlpy/result_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,8 @@
)

from psqlpy import ConnectionPool
from psqlpy.exceptions import RustPSQLDriverPyBaseError
from psqlpy._internal.exceptions import InterfaceError
from psqlpy.exceptions import DatabaseError
from taskiq import AsyncResultBackend, TaskiqResult
from taskiq.abc.serializer import TaskiqSerializer
from taskiq.compat import model_dump, model_validate
Expand Down Expand Up @@ -67,18 +68,20 @@ async def startup(self) -> None:
dsn=self.dsn,
**self.connect_kwargs,
)
await self._database_pool.execute(
querystring=CREATE_TABLE_QUERY.format(
self.table_name,
self.field_for_task_id,
),
)
await self._database_pool.execute(
querystring=CREATE_INDEX_QUERY.format(
self.table_name,
self.table_name,
),
)
async with self._database_pool.acquire() as connection:
await connection.execute(
querystring=CREATE_TABLE_QUERY.format(
self.table_name,
self.field_for_task_id,
),
)
async with self._database_pool.acquire() as connection:
await connection.execute(
querystring=CREATE_INDEX_QUERY.format(
self.table_name,
self.table_name,
),
)

async def shutdown(self) -> None:
"""Close the connection pool."""
Expand All @@ -94,15 +97,16 @@ async def set_result(
:param task_id: ID of the task.
:param result: result of the task.
"""
await self._database_pool.execute(
querystring=INSERT_RESULT_QUERY.format(
self.table_name,
),
parameters=[
task_id,
self.serializer.dumpb(model_dump(result)),
],
)
async with self._database_pool.acquire() as connection:
await connection.execute(
querystring=INSERT_RESULT_QUERY.format(
self.table_name,
),
parameters=[
task_id,
self.serializer.dumpb(model_dump(result)),
],
)

async def is_result_ready(self, task_id: str) -> bool:
"""Returns whether the result is ready.
Expand Down Expand Up @@ -135,26 +139,27 @@ async def get_result(
:raises ResultIsMissingError: if there is no result when trying to get it.
:return: TaskiqResult.
"""
connection: Final = await self._database_pool.connection()
try:
result_in_bytes: Final[bytes] = await connection.fetch_val(
querystring=SELECT_RESULT_QUERY.format(
self.table_name,
),
parameters=[task_id],
)
except RustPSQLDriverPyBaseError as exc:
async with self._database_pool.acquire() as connection:
result_in_bytes: Final[bytes] = await connection.fetch_val(
querystring=SELECT_RESULT_QUERY.format(
self.table_name,
),
parameters=[task_id],
)
except (DatabaseError, InterfaceError) as exc:
raise ResultIsMissingError(
f"Cannot find record with task_id = {task_id} in PostgreSQL",
) from exc

if not self.keep_results:
await self._database_pool.execute(
querystring=DELETE_RESULT_QUERY.format(
self.table_name,
),
parameters=[task_id],
)
async with self._database_pool.acquire() as conn:
await conn.execute(
querystring=DELETE_RESULT_QUERY.format(
self.table_name,
),
parameters=[task_id],
)

taskiq_result: Final = model_validate(
TaskiqResult[_ReturnType],
Expand Down
7 changes: 4 additions & 3 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,8 @@ async def psqlpy_result_backend(
)
await backend.startup()
yield backend
await backend._database_pool.execute(
querystring=f"DROP TABLE {postgres_table}",
)
async with backend._database_pool.acquire() as connection:
await connection.execute(
querystring=f"DROP TABLE {postgres_table}",
)
await backend.shutdown()
9 changes: 4 additions & 5 deletions tests/test_result_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,9 +89,10 @@ async def test_success_backend_default_result_delete_res(
with pytest.raises(expected_exception=ResultIsMissingError):
await backend.get_result(task_id=task_id)

await backend._database_pool.execute(
querystring=f"DROP TABLE {postgres_table}",
)
async with backend._database_pool.acquire() as connection:
await connection.execute(
querystring=f"DROP TABLE {postgres_table}",
)

await backend.shutdown()

Expand All @@ -106,7 +107,6 @@ async def test_success_backend_default_result(

:param default_taskiq_result: TaskiqResult with default result.
:param task_id: ID for task.
:param nats_urls: urls to NATS.
"""
await psqlpy_result_backend.set_result(
task_id=task_id,
Expand All @@ -127,7 +127,6 @@ async def test_success_backend_custom_result(

:param custom_taskiq_result: TaskiqResult with custom result.
:param task_id: ID for task.
:param redis_url: url to redis.
"""
await psqlpy_result_backend.set_result(
task_id=task_id,
Expand Down