beerds/server/modules/descriptions/repository/repository.py

85 lines
2.9 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from abc import ABCMeta, abstractmethod
from pathlib import Path
from aiocache import cached, Cache # type: ignore
from server.infra.db import AsyncDB
from server.modules.descriptions.domain import Breed
class ACharactersRepository(metaclass=ABCMeta):
@abstractmethod
async def get_characters(self) -> list[Breed]:
pass
async def get_character(self, alias: str) -> Breed | None:
pass
class CharactersRepository(ACharactersRepository):
def __init__(self):
pass
@cached(ttl=60, cache=Cache.MEMORY)
async def get_characters(self) -> list[Breed]:
breed_dir = Path("server/modules/descriptions/repository/breed_descriptions")
breeds: list[Breed] = []
# Идем по каждому текстовому файлу с описанием породы
for breed_file in breed_dir.glob("*.txt"):
breed_name = breed_file.stem # имя файла без расширения - название породы
description = breed_file.read_text(
encoding="utf-8"
) # читаем описание из файла
breeds.append(
Breed(
name=breed_name.replace("_", " "),
alias=breed_file.stem,
description=description.strip(),
)
)
breeds.sort(key=lambda b: b.name)
return breeds
async def get_character(self, alias: str) -> Breed | None:
breeds = await self.get_characters()
data = [b for b in breeds if b.alias == alias]
if len(data) == 0:
return None
return data[0]
class PGCharactersRepository(ACharactersRepository):
_db: AsyncDB
def __init__(self, db: AsyncDB):
self._db = db
@cached(ttl=60, cache=Cache.MEMORY)
async def get_characters(self) -> list[Breed]:
breed_dir = Path("server/modules/descriptions/repository/breed_descriptions")
breeds: list[Breed] = []
# Идем по каждому текстовому файлу с описанием породы
for breed_file in breed_dir.glob("*.txt"):
breed_name = breed_file.stem # имя файла без расширения - название породы
description = breed_file.read_text(
encoding="utf-8"
) # читаем описание из файла
breeds.append(
Breed(
name=breed_name.replace("_", " "),
alias=breed_file.stem,
description=description.strip(),
)
)
breeds.sort(key=lambda b: b.name)
return breeds
async def get_character(self, alias: str) -> Breed | None:
breeds = await self.get_characters()
data = [b for b in breeds if b.alias == alias]
if len(data) == 0:
return None
return data[0]