asyncpg storage (not done yet) by codefather-labs · Pull Request #1 · requestum-team/task-manager

codefather-labs · 2024-03-06T11:16:16Z

No description provided.

…k realizations

task_manager/storage/asyncpg_storage.py

stavichenko · 2024-03-07T10:03:12Z

task_manager/storage/asyncpg_storage.py

+        # todo postgres impl
+        self.on_task_callbacks.append(callback)
+
+        # self.subscribe_conn.add_listener(


stavichenko · 2024-03-12T07:54:00Z

task_manager/core/storage.py

        raise NotImplemented()
+
+
+class StorageType:


Core logic layer should not know about storage interface implementations. I would move this enum to cli command where it is used.

stavichenko · 2024-03-12T07:56:24Z

task_manager/storage/asyncpg_storage.py

+                notify_q = f"NOTIFY {self.subscription_channel}, '{idn}';"
+                await connection.execute(notify_q)
+
+        # FIXME what is this ?


stavichenko · 2024-03-12T07:58:12Z

task_manager/core/tasks.py

+
+
+@dataclass
+class Task(ConsumedTask):


no one method from storage interface use this class, so it shouldn't be presented here

replaced to task_manager.storage.tasks

stavichenko · 2024-03-12T08:01:31Z

task_manager/core/tasks.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass


if we want to move data types definitions to separate file, we should keep it in layer's package (task_manager.storage), for example we can consider next structure:

task_manager.storage.interface.py - Storage interface declaration

task_manager.datatypes.py - ConsumedTask and TransactionalResult declaration

task_manager._init_.py - import correspond items from subpackages

...or just keep it as it is in a single file for now

I think this will greatly disrupt the current structure. As I understand it, task_manager.core is a package of interfaces. I don’t see any point in transferring the storage interface from core to storages package, since this is an implementations package

stavichenko · 2024-03-12T08:03:25Z

task_manager/storage/asyncpg_storage.py

+
+
+@dataclass
+class AsyncPGTask(Task):


Not sure this class is required, can we just pass ConsumedTask to the TransactionalResult instance?

We need the fields status, error, description, which are not in ConsumedTask

This means we need to either edit ConsumedTask, which is not correct from the point of view of solid, or modify it - inherit from it and extend it. Which is what was done.

I removed AsyncPGTask and replaced it with Task from which it was inherited

stavichenko · 2024-03-12T08:12:01Z

task_manager/storage/asyncpg_storage.py

+
+        await self.pool.close() if self.pool else None
+
+    async def __create_transaction(self) -> Tuple[Transaction, Connection]:


why don't we use AsyncPG context manager isntead of this method?
async with connection.transaction()

Because some calls pass a transaction object along the stack. namely TransactionalResult has commit and rollback methods. In the AsyncPGConsumedTaskResult implementation, these methods commit or rollback transactions, respectively.

In this flow, if we use the context manager async with connection.transaction(), the transaction will be committed automatically immediately from the output of the context manager and the connection from the pool will be closed, even before calling the rollback and commit methods from AsyncPGConsumedTaskResult

task-manager/task_manager/storage/asyncpg_storage.py

Lines 31 to 49 in 8c8b6fd

class AsyncPGConsumedTaskResult(TransactionalResult[ConsumedTask]):

def __init__(

self,

task: AsyncPGTask,

lock: Transaction

):

self.task = task

self.lock = lock

async def get_data(self) -> AsyncPGTask:

# TODO another returning dataclass impl ?

# TODO I think current AsyncPGTask impl not bed

return self.task

async def commit(self):

await self.lock.commit()

async def rollback(self):

await self.lock.rollback()

we transfer control of the transaction object along the stack to the TaskManager

task-manager/task_manager/core/manager.py

Lines 28 to 43 in 8c8b6fd

async def take_pending_task(self, topics: list[str]):

if transaction := await self.storage.take_first_pending(topics):

await transaction.commit()

return await transaction.get_data()

return None

async def _on_new_task(self, idn: str):

if not (transaction := await self.storage.take_pending(idn)):

return

for session in self.sessions:

if session.deliver_task(await transaction.get_data()):

await transaction.commit()

return

await transaction.rollback()

that's why we don't use a context manager here - we need to transfer control of the open transaction along the stack to the TaskManager and it then decides to commit it or do a rollback

stavichenko · 2024-03-12T08:15:26Z

task_manager/storage/in_memory.py

-    status: int
-    error: None|str
-    description: str = ''
+from task_manager.core.storage import (


this file should bot be changed by this PR

Didn't understand. For what?
This is a ready-made InMemory implementation

stavichenko · 2024-03-12T08:17:02Z

test_postgres_storage.py

@@ -0,0 +1,86 @@
+import asyncio


Should it be removed? We already have a common test for both storage implementationsin this PR.

stavichenko · 2024-03-12T08:18:10Z

entrypoint.sh

@@ -0,0 +1,17 @@
+# DEPRECATED


I guess we can remove this entrypoint.

stavichenko · 2024-03-12T08:19:42Z

sqlmigrations/001_create_schema.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS tasks (


this file is related to postgresql storage implementation and should be placed in it's package resources folder.

codefather-labs added 6 commits March 6, 2024 13:12

not done yet

d2e6b45

not done yet

38e6fdc

pg impl conflict fix

4fbb0b6

easy refactoring

19ebc6d

create_task refactoring

d6b83f3

AsyncPGStorage refactoring + AsyncPGConsumedTaskResult and AsyncPGTas…

777e6b9

…k realizations

stavichenko reviewed Mar 7, 2024

View reviewed changes

postgres pub/sub, + refactoring

8c8b6fd

stavichenko reviewed Mar 12, 2024

View reviewed changes

pull requests updates

4fe6132


		await self.pool.close() if self.pool else None

		async def __create_transaction(self) -> Tuple[Transaction, Connection]:

	class AsyncPGConsumedTaskResult(TransactionalResult[ConsumedTask]):
	def __init__(
	self,
	task: AsyncPGTask,
	lock: Transaction
	):
	self.task = task
	self.lock = lock

	async def get_data(self) -> AsyncPGTask:
	# TODO another returning dataclass impl ?
	# TODO I think current AsyncPGTask impl not bed
	return self.task

	async def commit(self):
	await self.lock.commit()

	async def rollback(self):
	await self.lock.rollback()

	async def take_pending_task(self, topics: list[str]):
	if transaction := await self.storage.take_first_pending(topics):
	await transaction.commit()
	return await transaction.get_data()
	return None

	async def _on_new_task(self, idn: str):
	if not (transaction := await self.storage.take_pending(idn)):
	return

	for session in self.sessions:
	if session.deliver_task(await transaction.get_data()):
	await transaction.commit()
	return

	await transaction.rollback()

Conversation

codefather-labs commented Mar 6, 2024

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codefather-labs Mar 15, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codefather-labs Mar 15, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codefather-labs Mar 15, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

codefather-labs Mar 15, 2024 •

edited

Loading

codefather-labs Mar 15, 2024 •

edited

Loading

codefather-labs Mar 15, 2024 •

edited

Loading