Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions docs/quickstart.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
76 changes: 73 additions & 3 deletions src/hwpx/document.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import io
from datetime import datetime
import logging
import uuid
Expand Down Expand Up @@ -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
Expand All @@ -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":
Expand All @@ -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
Expand Down
65 changes: 65 additions & 0 deletions tests/test_document_context_manager.py
Original file line number Diff line number Diff line change
@@ -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