avnazv/toolbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Завершена работа со складами

Browse Source
This commit is contained in:
Macbook
2025-12-10 23:03:32 +03:00
parent 197a4d0b1e
commit fcbe25f7ec
14 changed files with 689 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/handlers/__pycache__/actions.cpython-313.pyc
BIN
View File
Binary file not shown.
db/handlers/__pycache__/stock.cpython-313.pyc
BIN
View File
Binary file not shown.
db/handlers/actions.py
+1 -1
View File
@@ -20,7 +20,7 @@ class StocksActions:
price: int,
placement: str,
reason: str,
):
) -> bool:
logger.info(f"Приход инструмента {toolkit_id} на складе {toolbox_id} ...")
newStocks = await StockHandler.add(
toolkit_id=toolkit_id,
db/handlers/stock.py
+92 -17
View File
@@ -1,34 +1,109 @@
from sqlalchemy import select
from db.schemas.stock import Stock
from db.schemas.stock import Placement, Stock
from utils import logger
def filterQuantity(stocksData, filtered):
async def filterQuantity(stocksData, filtered):
def filterStock(stock):
if stock.quantity > 0:
return stock
else:
return False
def convertPlacementData(placementList: list) -> dict:
placementDict = {}
for placementData in placementList:
toolboxId = placementData.toolbox_id
toolkitId = placementData.toolkit_id
placement = placementData.placement
if toolboxId not in placementDict:
placementDict[toolboxId] = {}
if toolkitId not in placementDict[toolboxId]:
placementDict[toolboxId][toolkitId] = placement
return placementDict
async def combinePlacementAndStock(stockData):
if stockData:
if isinstance(stockData, list):
placementList = await PlacementHandler.getAll()
placementDict = convertPlacementData(placementList)
for stock in stockData:
toolboxId = stock.get("toolbox_id")
toolkitId = stock.get("toolkit_id")
stock["placement"] = placementDict.get(toolboxId, {}).get(
toolkitId, None
)
else:
toolboxId = stockData.get("toolbox_id")
toolkitId = stockData.get("toolkit_id")
placement = await PlacementHandler.get(toolboxId, toolkitId)
stockData["placement"] = placement.placement
return stockData
exitData = []
if isinstance(stocksData, list):
if len(stocksData) == 0:
return []
stocksData.sort(key=lambda stock: stock.created_at)
filteredStocks = (
list(filter(filterStock, stocksData)) if filtered else stocksData
)
return [stock.toDict() for stock in filteredStocks] if filteredStocks else []
if len(stocksData) > 0:
stocksData.sort(key=lambda stock: stock.created_at)
filteredStocks = (
list(filter(filterStock, stocksData)) if filtered else stocksData
)
if filteredStocks:
exitData = [stock.toDict() for stock in filteredStocks]
else:
stock = filterStock(stocksData) if filtered else stocksData
if stock:
return stock.toDict()
exitData = stock.toDict()
return await combinePlacementAndStock(exitData)
class PlacementHandler:
async def add(**kwargs):
placement = kwargs.get("placement", None)
if not placement:
return
toolboxId = kwargs.get("toolbox_id", None)
toolkitId = kwargs.get("toolkit_id", None)
if not (toolkitId and toolboxId):
logger.error("toolkit_id or toolbox_id not found")
return
chechExists = await PlacementHandler.get(toolboxId, toolkitId)
if chechExists:
if placement != chechExists.placement:
logger.info(
f"Обновление места хранения инструмента {toolkitId} в ящике {toolboxId} с {chechExists.placement} на {placement}"
)
return await PlacementHandler.edit(chechExists, placement)
else:
return {}
logger.info(
f"Новое место хранения инструмента {toolkitId} в ящике {toolboxId} успешно создано {placement}"
)
return await Placement(**kwargs).save()
async def get(toolboxId: int, toolkitId: int) -> Placement:
from db import CRUD
return await CRUD.read(
select(Placement)
.where(Placement.toolbox_id == toolboxId)
.where(Placement.toolkit_id == toolkitId)
)
async def getAll() -> list[Placement]:
from db import CRUD
return await CRUD.read(select(Placement), all=True)
async def edit(toolboxId: int, toolkitId: int, placement: str):
placementDB = await PlacementHandler.get(toolboxId, toolkitId)
if placementDB and placementDB.placement != placement:
return await placementDB.edit(placement)
class StockHandler:
async def add(**kwargs) -> dict:
newStock = await Stock(**kwargs).save()
if "placement" in kwargs and kwargs["placement"]:
await PlacementHandler.add(**kwargs)
logger.info(
f"Новая запись об инструменте {newStock.toolkit_data.title} на складе {newStock.toolbox_data.title} успешно создана"
)
@@ -38,7 +113,7 @@ class StockHandler:
from db import CRUD
stocks = await CRUD.read(select(Stock), all=True)
return filterQuantity(stocks, filtered)
return await filterQuantity(stocks, filtered)
async def get(stockId: int, filtered: bool = True, record: bool = False):
from db import CRUD
@@ -47,21 +122,21 @@ class StockHandler:
if not stock:
logger.error(f"Запись {stockId} об остатках не найдена")
return {}
return filterQuantity(stock, filtered) if not record else stock
return await filterQuantity(stock, filtered) if not record else stock
async def getByToolboxId(toolboxId: int, filtered: bool = True):
from db import CRUD
query = select(Stock).where(Stock.toolbox_id == toolboxId)
stocks = await CRUD.read(query, True)
return filterQuantity(stocks, filtered)
return await filterQuantity(stocks, filtered)
async def getByToolkitId(toolkitId: int, filtered: bool = True):
from db import CRUD
query = select(Stock).where(Stock.toolkit_id == toolkitId)
stocks = await CRUD.read(query, True)
return filterQuantity(stocks, filtered)
return await filterQuantity(stocks, filtered)
async def getByToolboxIdAndToolkitId(
toolboxId: int, toolkitId: int, filtered: bool = True
@@ -74,7 +149,7 @@ class StockHandler:
.where(Stock.toolkit_id == toolkitId)
)
stocks = await CRUD.read(query, True)
return filterQuantity(stocks, filtered)
return await filterQuantity(stocks, filtered)
async def getByToolboxIdAndToolkitIdAndQPrice(
toolboxId: int, toolkitId: int, price: float, filtered: bool = True
@@ -88,7 +163,7 @@ class StockHandler:
.where(Stock.price == price)
)
stocks = await CRUD.read(query, True)
return filterQuantity(stocks, filtered)
return await filterQuantity(stocks, filtered)
async def edit(stockId: int, **kwargs) -> dict:
from db import CRUD
db/schemas/__pycache__/stock.cpython-313.pyc
BIN
View File
Binary file not shown.
db/schemas/stock.py
+35 -2
View File
@@ -32,13 +32,15 @@ class Stock(Base):
)
quantity = Column(Integer, nullable=False)
price = Column(Float, nullable=False)
placement = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
try:
setattr(self, key, value)
except Exception:
pass
def toDict(self):
return utils.toDict(self)
@@ -48,3 +50,34 @@ class Stock(Base):
async def edit(self, **kwargs):
return await CRUD.update(Stock, self.id, **kwargs)
class Placement(Base):
__tablename__ = "placements"
id = Column(Integer, primary_key=True, index=True)
toolbox_id = Column(
Integer, ForeignKey("toolboxes.id", ondelete="CASCADE"), nullable=False
)
toolkit_id = Column(
Integer, ForeignKey("toolkits.id", ondelete="CASCADE"), nullable=False
)
placement = Column(String, nullable=False)
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
def __init__(self, **kwargs):
for key, value in kwargs.items():
try:
setattr(self, key, value)
except Exception:
pass
def toDict(self):
return utils.toDict(self)
async def save(self) -> "Placement":
return await CRUD.create(self, refresh=True)
async def edit(self, placement: str):
return await CRUD.update(Placement, self.id, placement=placement)