diff --git a/docs/quickstart.md b/docs/quickstart.md index 292b401..a530e0d 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -30,6 +30,16 @@ print("첫 번째 문단 텍스트:", document.paragraphs[0].text) `HwpxDocument.open()`은 파일 경로, 바이트, 파일 객체 등 다양한 입력을 받아 문서를 로드합니다. 반환된 `document` 객체는 섹션, 문단, 표 등 주요 구성 요소에 바로 접근할 수 있는 고수준 API를 제공합니다. +컨텍스트 매니저(`with`)와 함께 사용하면 블록 종료 시점(정상/예외 모두)에도 내부 자원 정리가 자동으로 수행됩니다. + +```python +from hwpx import HwpxDocument + +with HwpxDocument.open("input/sample.hwpx") as document: + document.add_paragraph("with 블록 안에서 안전하게 편집") + document.save("output/sample-updated.hwpx") +``` + ## 2. 새 문단 추가하기 ```python diff --git a/src/hwpx/document.py b/src/hwpx/document.py index d20b438..66b8a80 100644 --- a/src/hwpx/document.py +++ b/src/hwpx/document.py @@ -2,6 +2,7 @@ from __future__ import annotations +import io from datetime import datetime import logging import uuid @@ -59,9 +60,17 @@ def _append_element( class HwpxDocument: """Provides a user-friendly API for editing HWPX documents.""" - def __init__(self, package: HwpxPackage, root: HwpxOxmlDocument): + def __init__( + self, + package: HwpxPackage, + root: HwpxOxmlDocument, + *, + managed_resources: tuple[Any, ...] = (), + ): self._package = package self._root = root + self._managed_resources = list(managed_resources) + self._closed = False # ------------------------------------------------------------------ # construction helpers @@ -76,9 +85,15 @@ def open( HwpxStructureError: 필수 파일이나 구조가 올바르지 않은 HWPX를 열 때 발생합니다. HwpxPackageError: 패키지를 여는 과정에서 일반적인 I/O/포맷 오류가 발생하면 전달됩니다. """ - package = HwpxPackage.open(source) + internal_resources: list[Any] = [] + open_source = source + if isinstance(source, bytes): + stream = io.BytesIO(source) + open_source = stream + internal_resources.append(stream) + package = HwpxPackage.open(open_source) root = HwpxOxmlDocument.from_package(package) - return cls(package, root) + return cls(package, root, managed_resources=tuple(internal_resources)) @classmethod def new(cls) -> "HwpxDocument": @@ -96,6 +111,61 @@ def from_package(cls, package: HwpxPackage) -> "HwpxDocument": root = HwpxOxmlDocument.from_package(package) return cls(package, root) + def __enter__(self) -> "HwpxDocument": + """컨텍스트 매니저 진입 시 현재 문서 인스턴스를 반환합니다.""" + + return self + + def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> bool: + """예외 발생 여부와 무관하게 내부 자원을 안전하게 정리합니다.""" + + self.close() + return False + + def close(self) -> None: + """문서가 관리하는 내부 패키지/스트림 자원을 정리합니다. + + 정리 정책: + - ``flush()`` 가능한 자원은 먼저 flush를 시도합니다. + - ``close()`` 가능한 자원은 flush 이후 close를 시도합니다. + - flush/close 중 발생한 예외는 로깅하고 무시하여 정리 루틴을 계속 진행합니다. + - 같은 문서에서 ``close()``를 여러 번 호출해도 안전합니다. + """ + + if self._closed: + return + + self._flush_resource(self._package) + for resource in self._managed_resources: + self._flush_resource(resource) + + self._close_resource(self._package) + for resource in self._managed_resources: + self._close_resource(resource) + + self._managed_resources.clear() + self._closed = True + + @staticmethod + def _flush_resource(resource: Any) -> None: + flush = getattr(resource, "flush", None) + if not callable(flush): + return + try: + flush() + except Exception: + logger.debug("자원 flush 중 예외를 무시합니다: resource=%r", resource, exc_info=True) + + @staticmethod + def _close_resource(resource: Any) -> None: + close = getattr(resource, "close", None) + if not callable(close): + return + try: + close() + except Exception: + logger.debug("자원 close 중 예외를 무시합니다: resource=%r", resource, exc_info=True) + # ------------------------------------------------------------------ # properties exposing document content @property diff --git a/tests/test_document_context_manager.py b/tests/test_document_context_manager.py new file mode 100644 index 0000000..57bc814 --- /dev/null +++ b/tests/test_document_context_manager.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import pytest + +from hwpx.document import HwpxDocument +from hwpx.templates import blank_document_bytes + + +class _TrackingResource: + def __init__(self, *, flush_error: bool = False, close_error: bool = False) -> None: + self.flush_calls = 0 + self.close_calls = 0 + self.flush_error = flush_error + self.close_error = close_error + + def flush(self) -> None: + self.flush_calls += 1 + if self.flush_error: + raise RuntimeError("flush failed") + + def close(self) -> None: + self.close_calls += 1 + if self.close_error: + raise RuntimeError("close failed") + + +def test_with_open_closes_internal_stream_when_exception_occurs() -> None: + internal_stream = None + + with pytest.raises(RuntimeError, match="boom"): + with HwpxDocument.open(blank_document_bytes()) as document: + assert document._managed_resources + internal_stream = document._managed_resources[0] + assert getattr(internal_stream, "closed", False) is False + raise RuntimeError("boom") + + assert internal_stream is not None + assert internal_stream.closed is True + + +def test_context_manager_flushes_and_closes_managed_resource() -> None: + document = HwpxDocument.new() + tracked = _TrackingResource() + document._managed_resources.append(tracked) + + with pytest.raises(ValueError, match="context"): + with document: + raise ValueError("context") + + assert tracked.flush_calls == 1 + assert tracked.close_calls == 1 + + +def test_close_ignores_resource_cleanup_errors_and_continues() -> None: + document = HwpxDocument.new() + broken = _TrackingResource(flush_error=True, close_error=True) + healthy = _TrackingResource() + document._managed_resources.extend([broken, healthy]) + + document.close() + + assert broken.flush_calls == 1 + assert broken.close_calls == 1 + assert healthy.flush_calls == 1 + assert healthy.close_calls == 1