44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
"""内存仓库。只给单测用，进程退出即丢。运行时走 Oracle。
|
||
|
||
按 oms_id 判重，已存在就 DuplicateOmsId，不覆盖。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from .repo import DuplicateFid, DuplicateOmsId
|
||
|
||
|
||
class MemorySalesGoodsRepo:
|
||
def __init__(self):
|
||
self._rows: dict[int, dict] = {}
|
||
|
||
def insert(self, record: dict) -> None:
|
||
oms_id = record["oms_id"]
|
||
if oms_id in self._rows:
|
||
raise DuplicateOmsId(oms_id)
|
||
self._rows[oms_id] = dict(record)
|
||
|
||
def get(self, oms_id: int) -> dict | None:
|
||
row = self._rows.get(oms_id)
|
||
return dict(row) if row is not None else None
|
||
|
||
def count(self) -> int:
|
||
return len(self._rows)
|
||
|
||
|
||
class MemoryPurchaseContractRepo:
|
||
def __init__(self):
|
||
self._rows: dict[int, dict] = {}
|
||
|
||
def insert(self, record: dict) -> None:
|
||
fid = record["fid"]
|
||
if fid in self._rows:
|
||
raise DuplicateFid(fid)
|
||
self._rows[fid] = dict(record)
|
||
|
||
def get(self, fid: int) -> dict | None:
|
||
row = self._rows.get(fid)
|
||
return dict(row) if row is not None else None
|
||
|
||
def count(self) -> int:
|
||
return len(self._rows)
|