db2-hj3415 2.4.3

MongoDB storage helpers for NFS (latest/snapshots)

db2-hj3415 README

db2-hj3415는 MongoDB 저장소 접근을 표준화한 라이브러리 패키지입니다.
NFS/Universe/Analysis/Prediction 데이터를 latest/snapshot 형태로 저장·조회할 수 있도록 공통 repo 함수를 제공합니다.

pyproject.toml 기준 정보:

• 패키지명: db2-hj3415
• 버전: 2.4.1
• Python: >=3.11
• 빌드 백엔드: flit_core
• 의존성: pymongo, contracts-hj3415, pydantic, pydantic-settings

1. 프로젝트 기능 설명

이 패키지는 다음 기능을 제공합니다.

• 비동기 Mongo 연결 래퍼: db2_hj3415.mongo.Mongo
• 환경설정 로더: db2_hj3415.settings.Settings, get_settings()
• 도메인별 저장소 함수:
  • db2_hj3415.nfs.repo
  • db2_hj3415.universe.repo
  • db2_hj3415.analysis.repo
  • db2_hj3415.prediction.repo
• 계약 DTO(contracts-hj3415)와 DB 문서 간 매퍼

즉, 서비스 코드에서는 “Mongo 연결 + repo 호출”만으로 저장/조회 로직을 구현할 수 있습니다.

2. 프로젝트 구조 설명

db2-hj3415/
├─ pyproject.toml
└─ src/db2_hj3415/
   ├─ settings.py         # 환경변수 로딩(Settings)
   ├─ mongo.py            # AsyncMongoClient 래퍼
   ├─ common/             # 공통 타입/유틸
   ├─ nfs/                # NFS repo/mappers/domain
   ├─ universe/           # Universe repo/mappers/domain
   ├─ analysis/           # Analysis repo/mappers/domain
   ├─ prediction/         # Prediction repo/domain
   └─ dart/               # DART 관련 repo/mappers/domain

사용 흐름:

1. get_settings()로 설정 로드
2. Mongo(settings) 생성 후 get_db()
3. 원하는 도메인 repo 함수 호출
4. 종료 시 await mongo.close()

3. 다른 프로젝트에서 임포트해서 사용하는 예시

설치:

python -m pip install db2-hj3415

기본 연결 예시:

import asyncio

from db2_hj3415.mongo import Mongo
from db2_hj3415.settings import get_settings


async def main() -> None:
    mongo = Mongo(get_settings())
    try:
        db = mongo.get_db()
        print(await db.command({"ping": 1}))
    finally:
        await mongo.close()


asyncio.run(main())

NFS latest 조회 예시:

import asyncio

from db2_hj3415.mongo import Mongo
from db2_hj3415.nfs.repo import get_latest
from db2_hj3415.settings import get_settings


async def main() -> None:
    mongo = Mongo(get_settings())
    try:
        db = mongo.get_db()
        doc = await get_latest(db, endpoint="c101", code="005930")
        print(doc)
    finally:
        await mongo.close()


asyncio.run(main())

Universe 코드 목록 조회 예시:

import asyncio

from db2_hj3415.mongo import Mongo
from db2_hj3415.settings import get_settings
from db2_hj3415.universe.repo import list_latest_universe_codes


async def main() -> None:
    mongo = Mongo(get_settings())
    try:
        db = mongo.get_db()
        codes = await list_latest_universe_codes(db, universe="krx300")
        print(codes[:10])
    finally:
        await mongo.close()


asyncio.run(main())

4. 환경변수 설정 예시

db2_hj3415.settings.Settings는 기본적으로 현재 작업 디렉터리의 .env와 OS 환경변수를 읽습니다.

주요 환경변수:

• MONGO_URI: MongoDB URI
• DB_NAME: DB 이름
• MONGO_CONNECT_TIMEOUT_MS: 연결 타임아웃(ms)
• MONGO_SERVER_SELECTION_TIMEOUT_MS: 서버 선택 타임아웃(ms)
• SNAPSHOT_TTL_DAYS: snapshot TTL 일수 (0 또는 None이면 TTL 비활성화)

예시:

MONGO_URI=mongodb://localhost:27017
DB_NAME=hj3415
MONGO_CONNECT_TIMEOUT_MS=5000
MONGO_SERVER_SELECTION_TIMEOUT_MS=5000
SNAPSHOT_TTL_DAYS=730

운영 팁:

• 앱 시작 시 인덱스 보장 함수(ensure_indexes)를 한 번 실행하는 구성이 안전합니다.
• 이 패키지는 비동기 API 중심이므로 asyncio.run(...) 또는 상위 async 컨텍스트에서 호출해야 합니다.