직접 낙찰(담당자 오프라인 계약가)을 AI 자동 낙찰과 통계·화면에서 구분하기 위해
낙찰 방식을 전용 컬럼으로 박고, 계약가·사유를 성격대로 배치했다. 처음엔 계약가·사유를
sessions.custom.offline_award(JSONB) 한 뭉치에 넣었으나 성격이 갈려 정리한다.
DB (postgres-init: init.sql 정본 + alters/2026-08-11-award-type.sql 보정, 멱등)
- quotations.award_type SMALLINT — 낙찰 방식(1=자동/2=직접). 통계 조회·집계 축이라 컬럼
- quotations.custom JSONB — 직접 낙찰 사유·처리자·시각(award={reason,by,at}). 표시·감사용
- sessions.contract_price BIGINT — 직접 낙찰 계약가. bid_price/reject_price 와 같은 협력사
가격 축이라 세션에. 자동 낙찰은 NULL(투찰가가 곧 계약가)
- 기존 offline_award(JSONB) 데이터를 컬럼·견적 custom 으로 이관 후 키 제거
backend
- 자동 낙찰(close_and_decide)=AUTO, 직접 낙찰(claim_for_award)=MANUAL 로 award_type 기록
- 직접 낙찰: 계약가→세션 contract_price, 사유·처리자·시각→견적 custom.award (한 트랜잭션)
- 통계 계약가 = coalesce(contract_price, bid_price) — JSONB 캐스팅 제거(컬럼끼리, 인덱스·타입 안전)
- QuotationData 에 award_type·custom, SessionData 에 contract_price 노출
- 안 쓰게 된 merge_session_custom 제거
front
- 견적 상세 결과 밴드에 '직접 낙찰' 배지 + 낙찰 사유 표시(견적 custom.award.reason).
거부·미참여 낙찰은 이미 직접 낙찰을 함의하므로 '— 낙찰' 꼬리를 떼 중복 표기 제거
- 목록 결과 배지에 '낙찰(직접)' 표기 — AI 자동낙찰과 한눈에 구분
- offlineAward()→directAwardPrice()/awardMeta() 로 교체(세션 컬럼·견적 custom 에서 읽음)
- 협상현황 표: 부가정보(회사 필드)와 의견(custom.opinion)을 별도 컬럼으로 분리
- 미응찰 건 레일 4번 칸 라벨 '최저 투찰가'→'결과'(투찰 없을 때)
테스트: negodata 110건 통과. 프론트 tsc+eslint 통과. dev DB 적용·화면 확인.
|
||
|---|---|---|
| .. | ||
| common | ||
| config | ||
| crud | ||
| docs | ||
| loadtest | ||
| router | ||
| scheduler | ||
| scripts | ||
| services | ||
| tests | ||
| .dockerignore | ||
| conftest.py | ||
| Dockerfile | ||
| pytest.ini | ||
| README.md | ||
| requirements.txt | ||
| web_main.py | ||
Negodata Backend
DerbyMasters_Server 아키텍처를 이식한 FastAPI 백엔드. 인증(JWT) 위에 견적·협력사·상품·견적설정·대시보드·알림·협상카드·회사유저관리 도메인과 견적 마감 스케줄러를 구현. negosium-backend 와 동일 구조이며, 실행·테스트·벤치마크 종합은 레포 최상위 README 참고.
디렉토리 구조
negodata/backend/
├── web_main.py # 엔트리포인트 (uvicorn)
├── config/ # 환경설정 (APP_ENV 별 toml 로드)
├── conftest.py, tests/ # pytest (test DB 자동 create/drop) — 아래 '테스트'
├── common/
│ ├── enums.py # ErrorType / 코드값 enum / EXCEPTION_*
│ ├── models/gmodel.py # 프로토콜 베이스 (WebPacketProtocol 등)
│ └── database/
│ ├── db_session_manager.py# ★ DB Read/Write + 람다 실행 핵심
│ └── model/models.py # ORM 모델 (companies·users·quotations·sessions·items·suppliers·notifications·cards …)
├── crud/ # 도메인별 DB 접근(I*CRUD 인터페이스+구현): quotation·supplier·item·dashboard·notification·card·user …
├── services/ # 비즈니스 로직: quotation·supplier·item·dashboard·notification·company_user·auth·email …
├── scheduler/ # 견적 마감 크론 잡(만료 마감 · 협상종결 마감)
└── router/
├── router.py # FastAPI app (CORS 등)
└── v1/ # 도메인별 라우터: auth·quotation·quotation_setting·supplier·item·card·dashboard·notification·company
└── validator/dependencies.py # ★ JWT 발급·검증, 해시(bcrypt), RemoveNoneResponse, RequireOwner
핵심 패턴
- MVC: router(컨트롤러) → service(로직) → crud(쿼리). crud 는 인터페이스/구현 분리.
- Depends 주입: service 가
IUserCRUD = Depends(UserCRUD), router 가Depends(AuthService), 인증은Depends(IsValidAccessToken). - DB Read/Write 분리:
DBType×DBWRType조합마다 별도 async 엔진(조회=Read, 변경=Write). - 람다 DB 실행: service 는 세션을 직접 열지 않고 람다를 매니저에 넘긴다. 세션/트랜잭션은 매니저가 책임.
err, acc = await DB_SESSION_MNG.execute_lambda( # 단일 조회 tbl_account.DBType(), DBWRType.DB_READ.value, lambda s: crud.get_account_by_id(s, id)) err = await DB_SESSION_MNG.execute_lambda_run( # 여러 변경 = 1 트랜잭션 [tbl_account.DBType()], [lambda s: crud.update_last_login(s, uid)]) - 비동기: 전 계층 async/await, SQLAlchemy async + asyncpg, task 단위 scoped session.
- Protocol 규약: 모든 패킷
WebPacketProtocol상속,Req_*/Res_*(응답엔result: ErrorInfo), 라우터별protocol.py. - ResponseNone: 응답의
None필드 재귀 제거(RemoveNoneResponse). - bcrypt 비차단:
GetHashedPW/VerifyPW를asyncio.to_thread로 오프로드(이벤트 루프 비차단). → 벤치마크
API 도메인 (/v1/*)
전체 스펙은 실행 후 http://localhost:9400/docs (Swagger). 주요 도메인:
| prefix | 요약 |
|---|---|
/v1/auth |
로그인 · access 토큰 재발급 · 내 정보(me). ※ 무인증 계정 생성은 제거됨 |
/v1/company/user |
최고관리자(OWNER) 전용 — 자기 회사 직원 계정 생성·관리 |
/v1/quotation |
견적 생성·목록·단건·마감·재생성 + 세션·채팅·낙찰결과·카드·초청메일 |
/v1/quotation-setting |
견적 설정(마진율 등) — 유저별 소유 |
/v1/supplier, /v1/item |
협력사 / 상품 CRUD (회사 스코프) |
/v1/card |
협상 카드 |
/v1/dashboard |
요약(회사 전체 + 내 견적) |
/v1/notification |
알림함(목록 · 읽음 처리) |
인증 헤더:
Authorization: Bearer <access_token>. 회사 소유 자원은 토큰의 회사로 스코프되고, 계정 관리는 OWNER 만 가능.
실행
cd negodata/backend
pip install -r requirements.txt
python web_main.py # APP_ENV 기본 local → http://localhost:9400/docs
환경: config.{local,test,prod}.toml (APP_ENV 로 선택).
테스트
테스트는 도커가 아니라 호스트(venv)에서 돌린다 — DB(PostgreSQL)만 도커(negosium-db, 127.0.0.1:5432)면 되고, 앱 컨테이너 안엔 pytest 가 없다. test DB(negosium_test_db)는 알아서 만들어졌다 지워지므로 수동 세팅이 필요 없다.
cd negodata/backend
python3 -m venv .venv && source .venv/bin/activate # 최초 1회 (venv 없을 때)
pip install -r requirements.txt # httpx 포함
pip install pytest pytest-asyncio # 테스트 도구(requirements 에 없음)
python -m pytest # 전체 (venv 활성화 상태)
python -m pytest -v # 테스트별 PASS/FAIL
python -m pytest tests/test_company_scope.py # 파일 하나만
python -m pytest -k scope # 이름에 'scope' 든 것만
venv 를 활성화(source .venv/bin/activate)하지 않으면 .venv/bin/python -m pytest 로 직접 지정한다.
(시스템에 python 명령이 없거나 pytest 가 venv 밖에 없으면 맨 python -m pytest 는 실패한다.)
정상이면 마지막 줄에 NN passed.
동작 방식 (전부 conftest.py 가 자동 처리 — 손댈 것 없음):
APP_ENV를test로 자동 설정 → config.test.toml 의negosium_test_db사용(dev DBnegosium_db와 완전 분리).- 세션 시작 시 test DB 를 새로 만들고(CREATE), 끝나면 내린다(DROP). 매번 현재 모델로 새로 빌드돼 스키마가 낡을 일이 없다. 남는 DB 도 없음.
- 테이블은
create_all로 자동 생성, 매 테스트 전TRUNCATE로 비워 격리. - 안전가드: 이름에
test없는 DB 는 만들지도 지우지도 않는다(실 DB 보호).
즉 새로 clone 받은 팀원도 Postgres 만 켜져 있으면
python -m pytest한 방이면 끝.