phzhik/chain-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Trying to fix thread leak: removed global dependencies

Browse Source
This commit is contained in:
Phil Zhitnikov 2024-08-01 23:49:03 +04:00
parent 8446d83896
commit d5a0cef68f
14 changed files with 85 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
chain_service/controllers/chain.py
View File

@ -1,15 +1,16 @@
from typing import Annotated
from loguru import logger
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Depends
from chain_service.database.models.chain import Chain
from chain_service.dependencies.chain_repository import ChainRepositoryDependency
from chain_service.dependencies.namespace_repository import (
NamespaceRepositoryDependency,
)
from chain_service.dependencies.file_uploader_service import (
FileUploaderServiceDependency,
)
from chain_service.dependencies.chain_repository import get_chain_repository
from chain_service.dependencies.namespace_repository import get_namespace_repository
from chain_service.dependencies.file_uploader_service import get_file_uploader_service
from chain_service.repositories.chain import ChainRepository
from chain_service.repositories.namespace import NamespaceRepository
from chain_service.services.file_uploader import FileUploaderService
router = APIRouter(prefix="/chain")
@ -17,9 +18,9 @@ router = APIRouter(prefix="/chain")
@router.post("/")
async def chain_upsert_controller(
chain: Chain,
chain_repository: ChainRepositoryDependency,
namespace_repository: NamespaceRepositoryDependency,
file_uploader_service: FileUploaderServiceDependency,
chain_repository: Annotated[ChainRepository, Depends(get_chain_repository)],
namespace_repository: Annotated[NamespaceRepository, Depends(get_namespace_repository)],
file_uploader_service: Annotated[FileUploaderService, Depends(get_file_uploader_service)],
):
try:
assert await namespace_repository.get_by_name(name=chain.namespace_id)
@ -39,8 +40,8 @@ async def chain_upsert_controller(
@router.get("/list")
async def chain_list_controller(
namespace_id: str,
chain_repository: ChainRepositoryDependency,
namespace_repository: NamespaceRepositoryDependency,
chain_repository: Annotated[ChainRepository, Depends(get_chain_repository)],
namespace_repository: Annotated[NamespaceRepository, Depends(get_namespace_repository)],
):
try:
@ -59,7 +60,7 @@ async def chain_list_controller(
@router.get("/{chain_id}")
async def chain_get_controller(
chain_id: str, chain_repository: ChainRepositoryDependency
chain_id: str, chain_repository: Annotated[ChainRepository, Depends(get_chain_repository)]
):
try:
assert (chain := await chain_repository.get_by_id(chain_id))
@ -76,7 +77,7 @@ async def chain_get_controller(
@router.delete("/delete/{chain_id}")
async def chain_delete_controller(
chain_id: str, chain_repository: ChainRepositoryDependency
chain_id: str, chain_repository: Annotated[ChainRepository, Depends(get_chain_repository)]
):
try:
await chain_repository.delete_by_id(chain_id)

13
chain_service/controllers/namespace.py
View File

@ -1,18 +1,19 @@
from typing import Annotated
from loguru import logger
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Depends
from chain_service.database.models.namespace import Namespace
from chain_service.dependencies.namespace_repository import (
NamespaceRepositoryDependency,
)
from chain_service.dependencies.namespace_repository import get_namespace_repository
from chain_service.repositories.namespace import NamespaceRepository
router = APIRouter(prefix="/namespace")
@router.post("/")
async def namespace_controller(
namespace: Namespace, namespace_repository: NamespaceRepositoryDependency
namespace: Namespace, namespace_repository: Annotated[NamespaceRepository, Depends(get_namespace_repository)]
):
try:
upserted_namespace = await namespace_repository.upsert(namespace)
@ -25,7 +26,7 @@ async def namespace_controller(
# @router.get("/{namespace_name}")
async def namespace_get_by_name_controller(
namespace_name: str, namespace_repository: NamespaceRepositoryDependency
namespace_name: str, namespace_repository: Annotated[NamespaceRepository, Depends(get_namespace_repository)]
):
try:
assert (namespace := await namespace_repository.get_by_name(namespace_name))

38
chain_service/controllers/run_chain.py
View File

@ -1,24 +1,22 @@
from typing import Annotated
from chain_service.repositories.chain import ChainRepository
from chain_service.repositories.progress_chain import ProgressChainRepository
from chain_service.repositories.running_chain import RunningChainRepository
from chain_service.schema.run_chain import RunChainInput, AbortChainInput
from chain_service.dependencies.chain_repository import ChainRepositoryDependency
from chain_service.dependencies.progress_chain_repository import (
ProgressChainRepositoryDependency,
)
from chain_service.dependencies.progress_chain_runner_service import (
ProgressChainRunnerServiceDependency,
)
from chain_service.dependencies.running_chain_repository import (
RunningChainRepositoryDependency,
)
from chain_service.dependencies.chain_repository import get_chain_repository
from chain_service.dependencies.progress_chain_repository import get_progress_chain_repository
from chain_service.dependencies.progress_chain_runner_service import get_progress_chain_runner_service
from chain_service.dependencies.running_chain_repository import get_running_chain_repository
from chain_service.database.models.progress_chain import ProgressChain
import asyncio
from loguru import logger
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Depends
from chain_service.services.progress_chain_runner import ProgressChainRunnerService
router = APIRouter()
@ -26,10 +24,10 @@ router = APIRouter()
@router.post("/run_chain")
async def run_chain_controller(
run_chain_input: RunChainInput,
chain_repository: ChainRepositoryDependency,
progress_chain_repository: ProgressChainRepositoryDependency,
progress_chain_runner_service: ProgressChainRunnerServiceDependency,
running_chain_repository: RunningChainRepositoryDependency,
chain_repository: Annotated[ChainRepository, Depends(get_chain_repository)],
progress_chain_repository: Annotated[ProgressChainRepository, Depends(get_progress_chain_repository)],
progress_chain_runner_service: Annotated[ProgressChainRunnerService, Depends(get_progress_chain_runner_service)],
running_chain_repository: Annotated[RunningChainRepository, Depends(get_running_chain_repository)],
):
try:
assert (chain := await chain_repository.get_by_id(run_chain_input.chain_id))
@ -70,7 +68,7 @@ async def run_chain_controller(
@router.post("/abort_chain")
async def abort_chain_controller(
abort_chain_input: AbortChainInput,
running_chain_repository: RunningChainRepositoryDependency,
running_chain_repository: Annotated[RunningChainRepository, Depends(get_running_chain_repository)],
):
try:
assert await running_chain_repository.exists(str(abort_chain_input.task_id))
@ -91,7 +89,7 @@ async def abort_chain_controller(
@router.post("/abort_all_chains")
async def abort_all_chains_controller(
running_chain_repository: RunningChainRepositoryDependency,
running_chain_repository: Annotated[RunningChainRepository, Depends(get_running_chain_repository)],
):
try:
await running_chain_repository.delete_all()

8
chain_service/dependencies/audio_converter_service.py
View File

@ -1,13 +1,5 @@
from chain_service.services.audio_converter import AudioConverterService
from fastapi import Depends
from typing import Annotated
def get_audio_converter_service() -> AudioConverterService:
return AudioConverterService()
AudioConverterServiceDependency = Annotated[
AudioConverterService, Depends(get_audio_converter_service)
]

7
chain_service/dependencies/chain_repository.py
View File

@ -1,12 +1,9 @@
from .database import DatabaseDependency
from .database import get_database, Database
from chain_service.repositories.chain import ChainRepository
from fastapi import Depends
from typing import Annotated
def get_chain_repository(database: DatabaseDependency) -> ChainRepository:
def get_chain_repository(database: Annotated[Database, Depends(get_database)]) -> ChainRepository:
return ChainRepository(database=database)
ChainRepositoryDependency = Annotated[ChainRepository, Depends(get_chain_repository)]

6
chain_service/dependencies/database.py
View File

@ -1,9 +1,6 @@
from chain_service.settings import Settings
from chain_service.database.database import Database
from fastapi import Depends
from typing import Annotated
def get_database() -> Database:
settings = Settings()
@ -11,6 +8,3 @@ def get_database() -> Database:
return Database(
database_url=settings.database_url, database_name=settings.database_name
)
DatabaseDependency = Annotated[Database, Depends(get_database)]

22
chain_service/dependencies/file_uploader_service.py
View File

@ -1,25 +1,25 @@
from planfix_client import PlanfixClient
from chain_service.services.file_uploader import FileUploaderService
from .planfix_client import PlanfixClientDependency
from .uploaded_file_repository import UploadedFileRepositoryDependency
from .audio_converter_service import AudioConverterServiceDependency
from .planfix_client import get_planfix_client
from .uploaded_file_repository import get_uploaded_file_repository
from .audio_converter_service import get_audio_converter_service
from fastapi import Depends
from typing import Annotated
from chain_service.repositories.uploaded_file import UploadedFileRepository
from chain_service.services.audio_converter import AudioConverterService
def get_file_uploader_service(
planfix_client: PlanfixClientDependency,
uploaded_file_repository: UploadedFileRepositoryDependency,
audio_converter_service: AudioConverterServiceDependency,
planfix_client: Annotated[PlanfixClient, Depends(get_planfix_client)],
uploaded_file_repository: Annotated[UploadedFileRepository, Depends(get_uploaded_file_repository)],
audio_converter_service: Annotated[AudioConverterService, Depends(get_audio_converter_service)],
) -> FileUploaderService:
return FileUploaderService(
planfix_client=planfix_client,
uploaded_file_repository=uploaded_file_repository,
audio_converter_service=audio_converter_service,
)
FileUploaderServiceDependency = Annotated[
FileUploaderService, Depends(get_file_uploader_service)
]

10
chain_service/dependencies/namespace_repository.py
View File

@ -1,15 +1,11 @@
from .database import DatabaseDependency
from chain_service.repositories.namespace import NamespaceRepository
from chain_service.database.database import Database
from .database import get_database
from fastapi import Depends
from typing import Annotated
def get_namespace_repository(database: DatabaseDependency) -> NamespaceRepository:
def get_namespace_repository(database: Annotated[Database, Depends(get_database)]) -> NamespaceRepository:
return NamespaceRepository(database=database)
NamespaceRepositoryDependency = Annotated[
NamespaceRepository, Depends(get_namespace_repository)
]

6
chain_service/dependencies/planfix_client.py
View File

@ -1,9 +1,6 @@
from planfix_client import PlanfixClient
from chain_service.settings import Settings
from fastapi import Depends
from typing import Annotated
def get_planfix_client() -> PlanfixClient:
settings = Settings()
@ -11,6 +8,3 @@ def get_planfix_client() -> PlanfixClient:
return PlanfixClient(
planfix_hostname=settings.planfix_hostname, planfix_token=settings.planfix_token
)
PlanfixClientDependency = Annotated[PlanfixClient, Depends(get_planfix_client)]

20
chain_service/dependencies/progress_action_service_factory.py
View File

@ -1,23 +1,21 @@
from .planfix_client import PlanfixClientDependency
from planfix_client import PlanfixClient
from .planfix_client import get_planfix_client
from chain_service.services.progress_action.factory import ProgressActionServiceFactory
from chain_service.dependencies.uploaded_file_repository import (
UploadedFileRepositoryDependency,
)
from chain_service.dependencies.uploaded_file_repository import get_uploaded_file_repository
from fastapi import Depends
from typing import Annotated
from chain_service.repositories.uploaded_file import UploadedFileRepository
def get_progress_action_service_factory(
planfix_client: PlanfixClientDependency,
uploaded_file_repository: UploadedFileRepositoryDependency,
planfix_client: Annotated[PlanfixClient, Depends(get_planfix_client)],
uploaded_file_repository: Annotated[UploadedFileRepository, Depends(get_uploaded_file_repository)]
) -> ProgressActionServiceFactory:
return ProgressActionServiceFactory(
planfix_client=planfix_client, uploaded_file_repository=uploaded_file_repository
)
ProgressActionServiceFactoryDependency = Annotated[
ProgressActionServiceFactory, Depends(get_progress_action_service_factory)
]

10
chain_service/dependencies/progress_chain_repository.py
View File

@ -1,16 +1,12 @@
from .database import DatabaseDependency
from chain_service.database.database import Database
from chain_service.repositories.progress_chain import ProgressChainRepository
from .database import get_database
from fastapi import Depends
from typing import Annotated
def get_progress_chain_repository(
database: DatabaseDependency,
database: Annotated[Database, Depends(get_database)],
) -> ProgressChainRepository:
return ProgressChainRepository(database=database)
ProgressChainRepositoryDependency = Annotated[
ProgressChainRepository, Depends(get_progress_chain_repository)
]

22
chain_service/dependencies/progress_chain_runner_service.py
View File

@ -1,26 +1,24 @@
from chain_service.services.progress_chain_runner import ProgressChainRunnerService
from .progress_chain_repository import ProgressChainRepositoryDependency
from .progress_action_service_factory import ProgressActionServiceFactoryDependency
from .running_chain_repository import RunningChainRepositoryDependency
from .progress_chain_repository import get_progress_chain_repository
from .progress_action_service_factory import get_progress_action_service_factory
from .running_chain_repository import get_running_chain_repository
from fastapi import Depends
from typing import Annotated
from chain_service.repositories.progress_chain import ProgressChainRepository
from chain_service.repositories.running_chain import RunningChainRepository
from chain_service.services.progress_action.factory import ProgressActionServiceFactory
def get_progress_chain_runner_service(
progress_chain_repository: ProgressChainRepositoryDependency,
progress_action_service_factory: ProgressActionServiceFactoryDependency,
running_chain_repository: RunningChainRepositoryDependency,
progress_chain_repository: Annotated[ProgressChainRepository, Depends(get_progress_chain_repository)],
progress_action_service_factory: Annotated[ProgressActionServiceFactory, Depends(get_progress_action_service_factory)],
running_chain_repository: Annotated[RunningChainRepository, Depends(get_running_chain_repository)],
) -> ProgressChainRunnerService:
return ProgressChainRunnerService(
progress_chain_repository=progress_chain_repository,
progress_action_service_factory=progress_action_service_factory,
running_chain_repository=running_chain_repository,
)
ProgressChainRunnerServiceDependency = Annotated[
ProgressChainRunnerService, Depends(get_progress_chain_runner_service)
]

10
chain_service/dependencies/running_chain_repository.py
View File

@ -1,4 +1,5 @@
from .database import DatabaseDependency
from .database import get_database
from chain_service.database.database import Database
from chain_service.repositories.running_chain import RunningChainRepository
from fastapi import Depends
@ -6,11 +7,6 @@ from typing import Annotated
def get_running_chain_repository(
database: DatabaseDependency,
database: Annotated[Database, Depends(get_database)],
) -> RunningChainRepository:
return RunningChainRepository(database=database)
RunningChainRepositoryDependency = Annotated[
RunningChainRepository, Depends(get_running_chain_repository)
]

9
chain_service/dependencies/uploaded_file_repository.py
View File

@ -1,16 +1,13 @@
from .database import DatabaseDependency
from chain_service.database.database import Database
from chain_service.repositories.uploaded_file import UploadedFileRepository
from .database import get_database
from fastapi import Depends
from typing import Annotated
def get_uploaded_file_repository(
database: DatabaseDependency,
database: Annotated[Database, Depends(get_database)],
) -> UploadedFileRepository:
return UploadedFileRepository(database=database)
UploadedFileRepositoryDependency = Annotated[
UploadedFileRepository, Depends(get_uploaded_file_repository)
]