feat: warehouse places accounting

.gitignore

@@ -1,4 +1,5 @@
 /migrations/versions/*
+/migrations_wms/versions/*
 .env
 .venv
 .idea
@@ -7,3 +8,4 @@ __pycache__
 /test
 /test/*
 certs
+.vscode

alembic.wms.ini

@@ -0,0 +1,114 @@
+# A generic, single database configuration.
+
+[alembic]
+# path to migration scripts
+script_location = migrations_wms
+
+# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
+# Uncomment the line below if you want the files to be prepended with date and time
+# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
+
+# sys.path path, will be prepended to sys.path if present.
+# defaults to the current working directory.
+prepend_sys_path = .
+
+# timezone to use when rendering the date within the migration file
+# as well as the filename.
+# If specified, requires the python>=3.9 or backports.zoneinfo library.
+# Any required deps can installed by adding `alembic[tz]` to the pip requirements
+# string value is passed to ZoneInfo()
+# leave blank for localtime
+# timezone =
+
+# max length of characters to apply to the
+# "slug" field
+# truncate_slug_length = 40
+
+# set to 'true' to run the environment during
+# the 'revision' command, regardless of autogenerate
+# revision_environment = false
+
+# set to 'true' to allow .pyc and .pyo files without
+# a source .py file to be detected as revisions in the
+# versions/ directory
+# sourceless = false
+
+# version location specification; This defaults
+# to migrations/versions.  When using multiple version
+# directories, initial revisions must be specified with --version-path.
+# The path separator used here should be the separator specified by "version_path_separator" below.
+# version_locations = %(here)s/bar:%(here)s/bat:migrations/versions
+
+# version path separator; As mentioned above, this is the character used to split
+# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
+# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
+# Valid values for version_path_separator are:
+#
+# version_path_separator = :
+# version_path_separator = ;
+# version_path_separator = space
+version_path_separator = os  # Use os.pathsep. Default configuration used for new projects.
+
+# set to 'true' to search source files recursively
+# in each "version_locations" directory
+# new in Alembic version 1.10
+# recursive_version_locations = false
+
+# the output encoding used when revision files
+# are written from script.py.mako
+# output_encoding = utf-8
+
+sqlalchemy.url = postgresql://{PG_LOGIN}:{PG_PASSWORD}@{PG_HOST}/{PG_DATABASE}
+
+
+[post_write_hooks]
+# post_write_hooks defines scripts or Python functions that are run
+# on newly generated revision scripts.  See the documentation for further
+# detail and examples
+
+# format using "black" - use the console_scripts runner, against the "black" entrypoint
+# hooks = black
+# black.type = console_scripts
+# black.entrypoint = black
+# black.options = -l 79 REVISION_SCRIPT_FILENAME
+
+# lint with attempts to fix using "ruff" - use the exec runner, execute a binary
+# hooks = ruff
+# ruff.type = exec
+# ruff.executable = %(here)s/.venv/bin/ruff
+# ruff.options = --fix REVISION_SCRIPT_FILENAME
+
+# Logging configuration
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

backend/config.py

@@ -4,13 +4,20 @@ from dotenv import load_dotenv
 
 load_dotenv()
 
-# Database
+# FF database
 PG_LOGIN = os.environ.get('PG_LOGIN')
 PG_PASSWORD = os.environ.get('PG_PASSWORD')
 PG_PORT = os.environ.get('PG_PORT')
 PG_DATABASE = os.environ.get('PG_DATABASE')
 PG_HOST = os.environ.get('PG_HOST')
 
+# WMS database
+WMS_PG_LOGIN = os.environ.get('WMS_PG_LOGIN')
+WMS_PG_PASSWORD = os.environ.get('WMS_PG_PASSWORD')
+WMS_PG_PORT = os.environ.get('WMS_PG_PORT')
+WMS_PG_DATABASE = os.environ.get('WMS_PG_DATABASE')
+WMS_PG_HOST = os.environ.get('WMS_PG_HOST')
+
 # Telegram
 TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN')
 

backend/dependecies.py

@@ -4,12 +4,14 @@ from fastapi import Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.session import get_session
+from backend.wms_session import get_wms_session
 from models import User
 from schemas.base import PaginationSchema
 from services.auth import get_current_user, authorized_user, guest_user
 from utils.dependecies import pagination_parameters
 
 SessionDependency = Annotated[AsyncSession, Depends(get_session)]
+WmsSessionDependency = Annotated[AsyncSession, Depends(get_wms_session)]
 PaginationDependency = Annotated[PaginationSchema, Depends(pagination_parameters)]
 CurrentUserDependency = Annotated[User, Depends(get_current_user)]
 AuthorizedUserDependency = Annotated[User, Depends(authorized_user)]

backend/wms_session.py

@@ -0,0 +1,21 @@
+from typing import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+
+from .config import WMS_PG_DATABASE, WMS_PG_HOST, WMS_PG_PASSWORD, WMS_PG_LOGIN
+
+DATABASE_URL = f'postgresql+asyncpg://{WMS_PG_LOGIN}:{WMS_PG_PASSWORD}@{WMS_PG_HOST}/{WMS_PG_DATABASE}'
+engine = create_async_engine(DATABASE_URL)
+wms_session_maker = sessionmaker(
+    engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+    autocommit=False,
+    autoflush=False,
+)
+
+
+async def get_wms_session() -> AsyncGenerator[AsyncSession, None]:
+    async with wms_session_maker() as session:
+        yield session

generators/base_pdf_card_generator/base_pdf_card_generator.py

@@ -63,3 +63,16 @@ class BasePdfCardGenerator:
             topMargin=1,
             bottomMargin=1
         )
+
+    def _get_paragraph_style(self, font_size: int):
+        common_paragraph_style = {
+            "parent": self.styles['Normal'],
+            "fontName": "DejaVuSans",
+            "spaceAfter": 4,
+            "fontSize": font_size,
+        }
+        return ParagraphStyle(
+            'LargeCentered',
+            alignment=1,
+            **common_paragraph_style,
+        )

generators/warehouse_place_qr_code_generator/__init__.py

@@ -0,0 +1 @@
+from .generator import WarehousePlaceQRCodeGenerator

generators/warehouse_place_qr_code_generator/generator.py

@@ -0,0 +1,46 @@
+from io import BytesIO
+
+from reportlab.lib.units import mm
+from reportlab.pdfgen.canvas import Canvas
+from reportlab.platypus import Paragraph, SimpleDocTemplate, Frame
+from reportlab_qrcode import QRCodeImage
+
+from generators.base_pdf_card_generator.base_pdf_card_generator import BasePdfCardGenerator
+from services.warehouse_management import WmsService
+
+
+class WarehousePlaceQRCodeGenerator(BasePdfCardGenerator):
+    async def generate(self, place_id: int, is_short: bool) -> BytesIO:
+        service = WmsService(self._session)
+        place_code = await service.get_code_of_place(place_id)
+
+        buffer = BytesIO()
+        doc: SimpleDocTemplate = self._create_doc(buffer)
+
+        def on_first_page(canvas: Canvas, doc):
+            denco_paragraph = Paragraph('DENCO', self.small_centered_style)
+
+            if is_short:
+                qr = QRCodeImage(place_code, size=20 * mm)
+                qr.drawOn(canvas, 39 * mm, -4)
+
+                number_paragraph = Paragraph(str(place_code), self._get_paragraph_style(26))
+                number_frame = Frame(x1=0 * mm, y1=-15 * mm, width=42 * mm, height=30 * mm)
+                denco_frame = Frame(x1=0 * mm, y1=-2 * mm, width=42 * mm, height=20 * mm)
+            else:
+                qr = QRCodeImage(place_code, size=25 * mm)
+                qr.drawOn(canvas, 34 * mm, -4)
+
+                number_paragraph = Paragraph(str(place_code), self._get_paragraph_style(34))
+                number_frame = Frame(x1=0 * mm, y1=9 * mm, width=60 * mm, height=30 * mm)
+                denco_frame = Frame(x1=0 * mm, y1=-13 * mm, width=58 * mm, height=20 * mm)
+
+            number_frame.addFromList([number_paragraph], canvas)
+            denco_frame.addFromList([denco_paragraph], canvas)
+
+        empty_paragraph = Paragraph("", self.small_centered_style)
+        elements = [empty_paragraph]
+        doc.build(elements, on_first_page)
+
+        buffer.seek(0)
+        return buffer

main.py

@@ -85,6 +85,7 @@ routers_list = [
     routers.status_router,
     routers.card_tag_router,
     routers.chat_router,
+    routers.wms_router,
 ]
 for router in routers_list:
     app.include_router(router)

migrate_wms.sh

@@ -0,0 +1,3 @@
+alembic -c alembic.wms.ini revision --autogenerate
+alembic -c alembic.wms.ini upgrade head
+

migrations_wms/README

@@ -0,0 +1 @@
+Generic single-database configuration with an async dbapi.

migrations_wms/env.py

@@ -0,0 +1,102 @@
+import asyncio
+import backend.config as settings
+from logging.config import fileConfig
+
+from sqlalchemy.engine import Connection
+
+from alembic import context
+
+from backend.wms_session import engine
+from models_wms import BaseModel
+
+# this is the Alembic Config object, which provides
+# access to the values within the .ini file in use.
+config = context.config
+
+# Interpret the config file for Python logging.
+# This line sets up loggers basically.
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+# add your model's MetaData object here
+# for 'autogenerate' support
+# from myapp import mymodel
+# target_metadata = mymodel.Base.metadata
+target_metadata = BaseModel.metadata
+
+
+# other values from the config, defined by the needs of env.py,
+# can be acquired:
+# my_important_option = config.get_main_option("my_important_option")
+# ... etc.
+def include_object(object, name, type_, reflected, compare_to):
+    return True  # Temporarily return True to debug all objects
+
+
+def get_url():
+    url = config.get_main_option("sqlalchemy.url").format(
+        PG_LOGIN=settings.PG_LOGIN,
+        PG_PASSWORD=settings.PG_PASSWORD,
+        PG_HOST=settings.PG_HOST,
+        PG_DATABASE=settings.PG_DATABASE,
+    )
+    return url
+
+
+def run_migrations_offline() -> None:
+    """Run migrations in 'offline' mode.
+
+    This configures the context with just a URL
+    and not an Engine, though an Engine is acceptable
+    here as well.  By skipping the Engine creation
+    we don't even need a DBAPI to be available.
+
+    Calls to context.execute() here emit the given string to the
+    script output.
+
+    """
+    url = get_url()
+    context.configure(
+        url=url,
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def do_run_migrations(connection: Connection) -> None:
+    context.configure(connection=connection,
+                      target_metadata=target_metadata,
+                      include_schemas=True,
+                      include_object=include_object,
+                      )
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+async def run_async_migrations() -> None:
+    """In this scenario we need to create an Engine
+    and associate a connection with the context.
+
+    """
+    connectable = engine
+
+    async with connectable.connect() as connection:
+        await connection.run_sync(do_run_migrations)
+
+    await connectable.dispose()
+
+
+def run_migrations_online() -> None:
+    """Run migrations in 'online' mode."""
+
+    asyncio.run(run_async_migrations())
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

migrations_wms/script.py.mako

@@ -0,0 +1,28 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, None] = ${repr(down_revision)}
+branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    ${downgrades if downgrades else "pass"}

models_wms/__init__.py

@@ -0,0 +1,5 @@
+from sqlalchemy.orm import configure_mappers
+
+from .place import *
+
+configure_mappers()

models_wms/base.py

@@ -0,0 +1,12 @@
+from sqlalchemy.ext.asyncio import AsyncAttrs
+from sqlalchemy.orm import declarative_base, DeclarativeBase
+
+
+class BaseModel(DeclarativeBase, AsyncAttrs):
+    def __repr__(self):
+        if hasattr(self, 'id'):
+            return f'<{self.__class__.__name__} id={self.id}>'
+        return super().__repr__()
+
+
+metadata = BaseModel.metadata

models_wms/place.py

@@ -0,0 +1,60 @@
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from models_wms.base import BaseModel
+
+
+class PlaceType(BaseModel):
+    __tablename__ = 'place_type'
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(nullable=False)
+    parent_id: Mapped[int] = mapped_column(
+        ForeignKey('place_type.id'),
+        nullable=True,
+    )
+    parent: Mapped['PlaceType'] = relationship(
+        'PlaceType',
+        lazy='noload',
+        back_populates='children',
+        primaryjoin='PlaceType.parent_id == PlaceType.id',
+        remote_side=[id],
+    )
+    children: Mapped[list['PlaceType']] = relationship(
+        'PlaceType',
+        lazy='selectin',
+        back_populates='parent',
+    )
+
+
+class Place(BaseModel):
+    __tablename__ = 'place'
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    number: Mapped[int] = mapped_column(nullable=False)
+
+    place_type_id: Mapped[int] = mapped_column(
+        ForeignKey('place_type.id'),
+        nullable=False,
+    )
+    place_type: Mapped[PlaceType] = relationship(
+        'PlaceType',
+        lazy='joined',
+    )
+
+    parent_id: Mapped[int] = mapped_column(
+        ForeignKey('place.id'),
+        nullable=True,
+    )
+    parent: Mapped['Place'] = relationship(
+        'Place',
+        lazy='noload',
+        back_populates='children',
+        primaryjoin='Place.parent_id == Place.id',
+        remote_side=[id],
+    )
+    children: Mapped[list['Place']] = relationship(
+        'Place',
+        lazy='selectin',
+        back_populates='parent',
+    )

routers/__init__.py

@@ -27,3 +27,4 @@ from .board import board_router
 from .status import status_router
 from .card_tag import card_tag_router
 from .chat import chat_router
+from .warehouse_management import wms_router

routers/warehouse_management.py

@@ -0,0 +1,150 @@
+from io import BytesIO
+
+from fastapi import APIRouter, Depends, Response
+
+from backend.dependecies import WmsSessionDependency
+from generators.warehouse_place_qr_code_generator import WarehousePlaceQRCodeGenerator
+from schemas.warehouse_management import *
+from services.auth import authorized_user
+from services.warehouse_management import WmsService
+
+wms_router = APIRouter(
+    prefix="/wms",
+    tags=["wms"],
+)
+
+
+# region Place types
+
+@wms_router.get(
+    '/place-type',
+    operation_id='get_place_types',
+    response_model=GetPlaceTypesResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def get_place_types(
+        session: WmsSessionDependency,
+):
+    return await WmsService(session).get_place_types()
+
+
+@wms_router.get(
+    '/place-type/flat/{parent_place_type_id}',
+    operation_id='get_flat_place_types',
+    response_model=GetFlatPlaceTypesResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def get_flat_place_types(
+        session: WmsSessionDependency,
+        parent_place_type_id: int,
+):
+    return await WmsService(session).get_flat_place_types(parent_place_type_id)
+
+
+@wms_router.post(
+    '/place-type',
+    operation_id='create_place_type',
+    response_model=CreatePlaceTypeResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def create_place_type(
+        session: WmsSessionDependency,
+        request: CreatePlaceTypeRequest,
+):
+    return await WmsService(session).create_place_type(request)
+
+
+@wms_router.patch(
+    '/place-type',
+    operation_id='edit_place_type',
+    response_model=EditPlaceTypeResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def edit_place_type(
+        session: WmsSessionDependency,
+        request: EditPlaceTypeRequest,
+):
+    return await WmsService(session).edit_place_type(request)
+
+
+@wms_router.delete(
+    '/place-type/{place_type_id}',
+    operation_id='delete_place_type',
+    response_model=DeletePlaceTypeResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def delete_place_type(
+        session: WmsSessionDependency,
+        place_type_id: int,
+):
+    return await WmsService(session).delete_place_type(place_type_id)
+
+
+# endregion
+
+# region Places
+
+@wms_router.get(
+    '/place',
+    operation_id='get_places',
+    response_model=GetPlacesResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def get_places(
+        session: WmsSessionDependency,
+):
+    return await WmsService(session).get_places()
+
+
+@wms_router.post(
+    '/place',
+    operation_id='create_place',
+    response_model=CreatePlaceResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def create_place(
+        session: WmsSessionDependency,
+        request: CreatePlaceRequest,
+):
+    return await WmsService(session).create_place(request)
+
+
+@wms_router.patch(
+    '/place',
+    operation_id='edit_place',
+    response_model=EditPlaceResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def edit_place(
+        session: WmsSessionDependency,
+        request: EditPlaceRequest,
+):
+    return await WmsService(session).edit_place(request)
+
+
+@wms_router.delete(
+    '/place/{place_id}',
+    operation_id='delete_place',
+    response_model=DeletePlaceResponse,
+    dependencies=[Depends(authorized_user)],
+)
+async def delete_place(
+        session: WmsSessionDependency,
+        place_id: int,
+):
+    return await WmsService(session).delete_place(place_id)
+
+
+@wms_router.get(
+    "/place/pdf/{place_id}/{is_short}",
+    operation_id="get_place_qr_code_pdf",
+)
+async def generate_place_qr_code_pdf(
+        session: WmsSessionDependency,
+        place_id: int,
+        is_short: bool = False,
+):
+    pdf_file: BytesIO = await WarehousePlaceQRCodeGenerator(session).generate(place_id, is_short)
+    return Response(pdf_file.getvalue(), media_type="application/pdf")
+
+# endregion

schemas/warehouse_management.py

@@ -0,0 +1,97 @@
+from typing import Optional
+
+from schemas.base import BaseSchema, OkMessageSchema
+
+
+# region Entities
+
+class BasePlaceTypeSchema(BaseSchema):
+    name: str
+    parent_id: Optional[int]
+
+
+class BasePlaceTypeWithCountSchema(BasePlaceTypeSchema):
+    child_count: int
+
+
+class FlatPlaceTypeSchema(BasePlaceTypeSchema):
+    id: int
+
+
+class PlaceTypeSchema(FlatPlaceTypeSchema):
+    children: list['PlaceTypeSchema'] = []
+    places_count: int
+
+
+class BasePlaceSchema(BaseSchema):
+    parent_id: Optional[int]
+    place_type_id: int
+
+
+class PlaceSchema(BasePlaceSchema):
+    id: int
+    number: int
+    children: list['PlaceSchema'] = []
+    place_type: BasePlaceTypeWithCountSchema
+
+
+# endregion
+
+# region Requests
+
+class EditPlaceTypeRequest(BaseSchema):
+    place_type: FlatPlaceTypeSchema
+
+
+class CreatePlaceTypeRequest(BaseSchema):
+    place_type: BasePlaceTypeSchema
+
+
+class EditPlaceRequest(BaseSchema):
+    place: PlaceSchema
+
+
+class CreatePlaceRequest(BaseSchema):
+    place: BasePlaceSchema
+
+
+# endregion
+
+# region Responses
+
+class GetPlaceTypesResponse(BaseSchema):
+    place_types: list[PlaceTypeSchema]
+
+
+class GetFlatPlaceTypesResponse(BaseSchema):
+    place_types: list[FlatPlaceTypeSchema]
+
+
+class CreatePlaceTypeResponse(OkMessageSchema):
+    pass
+
+
+class EditPlaceTypeResponse(OkMessageSchema):
+    pass
+
+
+class DeletePlaceTypeResponse(OkMessageSchema):
+    pass
+
+
+class GetPlacesResponse(BaseSchema):
+    places: list[PlaceSchema]
+
+
+class CreatePlaceResponse(OkMessageSchema):
+    pass
+
+
+class EditPlaceResponse(OkMessageSchema):
+    pass
+
+
+class DeletePlaceResponse(OkMessageSchema):
+    pass
+
+# endregion

services/warehouse_management.py

@@ -0,0 +1,222 @@
+from sqlalchemy import select, delete, func, literal
+from sqlalchemy.orm import aliased, selectinload
+
+from models_wms import PlaceType, Place
+from schemas.warehouse_management import *
+from services.base import BaseService
+
+
+class WmsService(BaseService):
+    @staticmethod
+    def create_places_from_list(nodes: list) -> list:
+        node_map = {node.id: node for node in nodes}
+        root_nodes = []
+
+        for node in node_map.values():
+            parent_id = node.parent_id
+            if parent_id is None:
+                root_nodes.append(node)
+            else:
+                parent = node_map.get(parent_id)
+                if parent:
+                    parent.children.append(node)
+
+        return root_nodes
+
+    async def get_place_types(self) -> GetPlaceTypesResponse:
+        pt_alias = aliased(PlaceType)
+
+        base_query = (
+            select(
+                PlaceType.id,
+                PlaceType.name,
+                PlaceType.parent_id,
+                func.count(Place.id).label("count")
+            )
+            .join(Place)
+            .group_by(PlaceType.id, PlaceType.parent_id, PlaceType.name)
+            .where(PlaceType.parent_id == None)
+        )
+
+        recursive_query = select(
+            pt_alias.id,
+            pt_alias.name,
+            pt_alias.parent_id,
+            literal(0).label("count")
+        ).join(
+            PlaceType,
+            PlaceType.id == pt_alias.parent_id,
+            isouter=True,
+        )
+
+        cte = base_query.union_all(recursive_query).cte(recursive=True)
+
+        stmt = select(cte)
+        result = await self.session.execute(stmt)
+
+        place_types = []
+        for type_id, name, parent_id, places_count in result:
+            place_type = PlaceTypeSchema(
+                id=type_id,
+                name=name,
+                children=[],
+                parent_id=parent_id,
+                places_count=places_count,
+            )
+            place_types.append(place_type)
+
+        place_types = WmsService.create_places_from_list(place_types)
+
+        return GetPlaceTypesResponse(place_types=place_types)
+
+    async def get_flat_place_types(self, parent_place_type_id: int) -> GetFlatPlaceTypesResponse:
+        stmt = select(PlaceType)
+        if parent_place_type_id != -1:
+            stmt = stmt.where(PlaceType.parent_id == parent_place_type_id)
+        else:
+            stmt = stmt.where(PlaceType.parent_id == None)
+
+        place_types = (await self.session.scalars(stmt)).all()
+
+        return GetFlatPlaceTypesResponse(place_types=place_types)
+
+    async def create_place_type(self, request: CreatePlaceTypeRequest) -> CreatePlaceTypeResponse:
+        place_type = PlaceType(
+            name=request.place_type.name,
+            parent_id=request.place_type.parent_id,
+        )
+        self.session.add(place_type)
+        await self.session.commit()
+        return CreatePlaceTypeResponse(ok=True, message='Тип места на складе успешно создан')
+
+    async def edit_place_type(self, request: EditPlaceTypeRequest) -> EditPlaceTypeResponse:
+        place_type = await self.session.get(PlaceType, request.place_type.id)
+        place_type.name = request.place_type.name
+        await self.session.commit()
+        return EditPlaceTypeResponse(ok=True, message='Тип места на складе успешно изменен')
+
+    async def delete_place_type(self, place_type_id: int) -> DeletePlaceTypeResponse:
+        stmt = (
+            delete(PlaceType)
+            .where(PlaceType.id == place_type_id)
+        )
+        await self.session.execute(stmt)
+        await self.session.commit()
+        return DeletePlaceTypeResponse(ok=True, message='Тип места на складе успешно удален')
+
+    async def get_raw_places(self) -> list[dict]:
+        pt_alias = aliased(Place)
+
+        base_query = select(Place)
+
+        recursive_query = (
+            select(pt_alias)
+            .join(
+                Place,
+                Place.id == pt_alias.parent_id
+            )
+        )
+
+        cte = base_query.union_all(recursive_query).cte(recursive=True)
+        recur_sub = select(cte).subquery()
+
+        place_type_child = aliased(PlaceType)
+
+        place_type_count_sub = (
+            select(func.count(place_type_child.id).label("child_count"))
+            .where(PlaceType.id == place_type_child.parent_id)
+            .scalar_subquery()
+        )
+        place_type_sub = (
+            select(PlaceType, place_type_count_sub.label("child_count"))
+            .order_by(PlaceType.id)
+            .subquery()
+        )
+
+        res_stmt = (
+            select(recur_sub, place_type_sub)
+            .join(place_type_sub, place_type_sub.c.id == recur_sub.c.place_type_id)
+        )
+
+        result = await self.session.execute(res_stmt)
+        result_dicts = result.mappings().all()
+        return [*result_dicts]
+
+    async def get_places(self) -> GetPlacesResponse:
+        result = await self.get_raw_places()
+
+        places = []
+        for place in result:
+            place_type = BasePlaceTypeWithCountSchema(
+                child_count=place['child_count'],
+                parent_id=place['parent_id_1'],
+                name=place['name'],
+            )
+            place = PlaceSchema(
+                id=place['id'],
+                number=place['number'],
+                children=[],
+                parent_id=place['parent_id'],
+                place_type=place_type,
+                place_type_id=place['place_type_id'],
+            )
+            places.append(place)
+
+        places = WmsService.create_places_from_list(places)
+
+        return GetPlacesResponse(places=places)
+
+    async def create_place(self, request: CreatePlaceRequest) -> CreatePlaceResponse:
+        stmt = (
+            select(func.max(Place.number))
+            .where(Place.parent_id == request.place.parent_id)
+        )
+        last_number = await self.session.scalar(stmt)
+        last_number = last_number if last_number else 0
+
+        place = Place(
+            number=last_number + 1,
+            parent_id=request.place.parent_id,
+            place_type_id=request.place.place_type_id,
+        )
+        self.session.add(place)
+        await self.session.commit()
+        return CreatePlaceResponse(ok=True, message='Место на складе успешно создано')
+
+    async def edit_place(self, request: EditPlaceRequest) -> EditPlaceResponse:
+        place = await self.session.get(Place, request.place.id)
+        place.name = request.place.name
+        await self.session.commit()
+        return EditPlaceResponse(ok=True, message='Место на складе успешно изменено')
+
+    async def delete_place(self, place_id: int) -> DeletePlaceResponse:
+        stmt = (
+            delete(Place)
+            .where(Place.id == place_id)
+        )
+        await self.session.execute(stmt)
+        await self.session.commit()
+        return DeletePlaceResponse(ok=True, message='Место на складе успешно удалено')
+
+    async def get_code_of_place(self, place_id: int) -> str:
+        select_depth = 10
+        option = selectinload(Place.parent)
+        for _ in range(select_depth):
+            option = option.selectinload(Place.parent)
+        stmt = (
+            select(Place)
+            .where(Place.id == place_id)
+            .options(option)
+        )
+        places = (await self.session.scalars(stmt)).all()
+
+        numbers: list[str] = []
+        if len(places) == 0:
+            return ''
+
+        place = places[0]
+        while place:
+            numbers.append(str(place.number))
+            place = place.parent
+
+        return '.'.join(numbers[::-1])