Python에서 제공하는 "io.BytesIO" 클래스는 메모리 상의 이진 데이터를 파일 객체처럼 다룰 수 있도록 도와줍니다.
특히 테스트 환경이나 네트워크 시뮬레이션에서 실제 파일을 만들지 않고도 손쉽게 데이터를 읽고 쓸 수 있다는 장점이 있습니다.
하지만 "BytesIO" 객체에서 데이터를 읽을 때, 단순히 "더 읽을 데이터가 남아 있는지" 확인하는 방법은 생각보다 직관적이지 않을 수 있습니다. 이번 글에서는 "BytesIO.read()"와 관련된 데이터 유무 확인 방법을 알아 보겠습니다.
1. "readable()" 메서드의 의미
먼저 많은 분들이 착각하실 수 있는 부분이 있습니다.
"BytesIO" 객체에는 "readable()" 메서드가 있는데, 이는 이 객체가 읽기 기능을 지원하는지 여부 만 반환합니다.
import io
buf = io.BytesIO(b"hello")
print(buf.readable()) # True
따라서 항상 "True"를 반환하며, 실제로 읽을 데이터가 남아 있는지와는 무관합니다.
즉, "readable()"은 데이터 유무를 확인하는 용도가 아닙니다.
2. "read()" 호출 시 동작
"BytesIO.read(size)" 메서드를 호출하면,
- 데이터가 남아 있다면 해당 크기만큼 반환
- 데이터가 더 이상 없다면 빈 바이트열("b''")을 반환합니다.
buf = io.BytesIO(b"abc")
print(buf.read(2)) # b'ab'
print(buf.read(2)) # b'c'
print(buf.read(2)) # b''
따라서 가장 단순한 방법은 "read()"의 결과가 빈 바이트열인지 확인하는 것입니다.
3. 현재 위치와 끝 위치 비교하기
조금 더 명확하게 "읽을 데이터가 남아 있는지" 확인하고 싶다면, 파일 포인터의 현재 위치와 전체 데이터의 길이를 비교하는 방법을 사용할 수 있습니다.
buf = io.BytesIO(b"abcdef")
# 현재 위치
pos = buf.tell()
# 전체 길이 확인
buf.seek(0, 2) # EOF로 이동
end = buf.tell()
# 다시 원래 위치 복원
buf.seek(pos)
print(pos < end) # True면 데이터가 남아 있음
이 방법을 이용하면, 실제 데이터를 읽지 않고도 남아 있는지 여부만 확인할 수 있습니다.
4. 비동기 코드에서의 활용
"BytesIO"는 기본적으로 동기적 객체이므로 "asyncio" 환경에서 직접 비동기적으로 동작하지는 않습니다.
다만, 비동기 코드 안에서 "BytesIO"의 데이터를 확인해야 할 경우, 위에서 살펴본 EOF(End Of File) 확인 방법을 활용할 수 있습니다.
import io
import asyncio
async def read_if_available(buf: io.BytesIO) -> bytes:
pos = buf.tell()
buf.seek(0, 2)
end = buf.tell()
buf.seek(pos)
if pos < end:
return buf.read()
return b"" # 데이터 없음
비동기 함수 안에서도 이러한 방식으로 안전하게 데이터를 확인할 수 있습니다.
5. 네트워크처럼 점진적으로 데이터가 들어오는 경우
만약 실제 파일이나 네트워크 소켓처럼 데이터가 나중에 들어올 수 있는 구조라면, "BytesIO"만으로는 이를 처리하기 어렵습니다. 이때는 "asyncio.StreamReader"나 "asyncio.Queue"를 활용하여 데이터가 도착할 때까지 대기한 후 읽는 방식을 사용하는 것이 적합합니다.
맺는말
정리하면, "BytesIO.readable()"은 단순히 "읽기 기능 지원 여부"를 의미할 뿐, 읽을 데이터가 남아 있는지 여부와는 관계가 없습니다.
읽을 데이터가 남아 있는지를 확인하려면,
1. "read()" 호출 후 반환값이 b''인지 확인
2. "tell()"과 "seek()"을 이용하여 현재 위치와 전체 길이를 비교
이 두 가지 방법을 활용할 수 있습니다.
비동기 코드에서 "BytesIO"를 사용할 때도 마찬가지로, 위 방식들을 적절히 조합하면 안정적으로 데이터를 확인하고 읽을 수 있습니다.
'프로그래밍 > Python' 카테고리의 다른 글
| [python] 단위시험을 위한 모의 클래스 작성 방법: 직접 상속 / 위임 (1) | 2025.10.17 |
|---|---|
| [python] "RuntimeWarning: coroutine 'AsyncMockMixin._execute_mock_call' was never awaited" 문제 (0) | 2025.10.15 |
| [python] "unittest"에서 "await"와 "run_until_complete()"의 차이점 (0) | 2025.09.17 |
| [python] pip 설치 중 SSL 인증서 오류 해결 방법 (0) | 2025.07.30 |
| [python] 네트워크 시험에서 Stub와 Driver의 개념과 예제 (0) | 2025.07.03 |