Python 파일/IO 완전 정리

Python code

from pathlib import Path
import json

# pathlib — 경로 조작
base = Path("/home/user/projects")
file = base / "data" / "config.json"   # / 연산자로 결합

print(file.name)        # config.json
print(file.stem)        # config
print(file.suffix)      # .json
print(file.parent)      # /home/user/projects/data

# 파일 읽기 (with문으로 자동 닫기)
sample = Path("sample.txt")
sample.write_text("Hello\nWorld\nPython", encoding="utf-8")

# 전체 읽기
content = sample.read_text(encoding="utf-8")

# 줄 단위 읽기
with sample.open(encoding="utf-8") as f:
    for line in f:                  # 메모리 효율적
        print(line.rstrip())

# 파일 쓰기
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("첫 번째 줄\n")
    f.writelines(["두\n", "세\n"])

# JSON 처리
data = {"name": "Alice", "scores": [95, 87, 92]}
Path("data.json").write_text(
    json.dumps(data, ensure_ascii=False, indent=2),
    encoding="utf-8"
)
loaded = json.loads(Path("data.json").read_text(encoding="utf-8"))

# 디렉토리 순회
for p in Path(".").glob("*.py"):
    print(p)

for p in Path(".").rglob("*.json"):  # 재귀
    print(p)

Python code

import csv, json, os
from pathlib import Path

# CSV 읽기
with open("data.csv", newline="", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row["name"], row["score"])

# CSV 쓰기
fields = ["name", "score", "grade"]
rows = [
    {"name": "Alice", "score": 95, "grade": "A"},
    {"name": "Bob",   "score": 82, "grade": "B"},
]
with open("output.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=fields)
    writer.writeheader()
    writer.writerows(rows)

# JSON — ensure_ascii=False 로 한글 보존
data = {"이름": "앨리스", "점수": 95}
json_str = json.dumps(data, ensure_ascii=False, indent=2)
parsed  = json.loads(json_str)

# 파일로 저장/로드
Path("data.json").write_text(json_str, encoding="utf-8")
with open("data.json", encoding="utf-8") as f:
    loaded = json.load(f)

# 환경 변수
from dotenv import load_dotenv   # pip install python-dotenv

load_dotenv()  # .env 파일 로드
db_url  = os.getenv("DATABASE_URL", "sqlite:///local.db")
api_key = os.environ.get("API_KEY")
port    = int(os.getenv("PORT", "8080"))

Python code

# === pyproject.toml 예제 ===
"""
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "my-awesome-lib"
version = "1.0.0"
description = "교육용 Python 패키지 예제"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
authors = [
    {name = "홍길동", email = "hong@example.com"},
]
dependencies = [
    "requests>=2.28",
    "pydantic>=2.0",
]

[project.optional-dependencies]
dev = ["pytest>=7.0", "ruff>=0.1.0", "mypy>=1.0"]
docs = ["sphinx>=7.0", "sphinx-rtd-theme"]

[project.scripts]
my-cli = "my_awesome_lib.cli:main"

[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "-v --tb=short"

[tool.ruff]
target-version = "py310"
line-length = 88

[tool.mypy]
strict = true
"""

# === 패키지 구조 ===
"""
my-awesome-lib/
├── pyproject.toml
├── README.md
├── src/
│   └── my_awesome_lib/
│       ├── __init__.py
│       ├── cli.py
│       └── core.py
└── tests/
    ├── __init__.py
    └── test_core.py
"""

# === 빌드 & 배포 명령어 ===
import subprocess, sys

commands = {
    "개발 모드 설치": "pip install -e '.[dev]'",
    "패키지 빌드":    "python -m build",
    "PyPI 업로드":   "python -m twine upload dist/*",
    "린트 검사":     "ruff check src/",
    "타입 검사":     "mypy src/",
    "테스트 실행":    "pytest",
}

print("패키징 워크플로우:")
for name, cmd in commands.items():
    print(f"  {name}: {cmd}")

# 실제 사용 예: 현재 환경의 패키지 정보
from importlib.metadata import version, requires
try:
    pip_ver = version("pip")
    print(f"\npip 버전: {pip_ver}")
except Exception:
    print("\npip 정보를 가져올 수 없습니다")

Python code

from pathlib import Path

# 경로 조합 (/ 연산자)
base = Path("/home/user")
config = base / "projects" / "app" / "config.yaml"
print(config)           # /home/user/projects/app/config.yaml
print(config.parent)    # /home/user/projects/app
print(config.stem)      # config
print(config.suffix)    # .yaml
print(config.name)      # config.yaml

# 경로 변환
print(config.with_suffix(".json"))  # .../config.json
print(config.with_name("settings.yaml"))

# 현재 디렉터리 기준
here = Path.cwd()
print(f"현재: {here}")

# 파일 검색 (glob)
# python_files = list(Path(".").rglob("*.py"))  # 재귀 검색
# for f in python_files[:5]:
#     print(f"  {f}: {f.stat().st_size} bytes")

# 파일 읽기/쓰기 (한 줄)
# Path("output.txt").write_text("Hello!", encoding="utf-8")
# content = Path("output.txt").read_text(encoding="utf-8")

# 경로 존재 확인
p = Path("/tmp/test")
print(f"존재: {p.exists()}")
print(f"파일: {p.is_file()}")
print(f"디렉터리: {p.is_dir()}")

# 부모 순회
for parent in config.parents:
    print(f"  상위: {parent}")

Python code

from pathlib import Path
import io

# 줄 단위 읽기 (메모리 효율적)
def count_lines(filepath: str) -> int:
    count = 0
    with open(filepath, "r", encoding="utf-8") as f:
        for _ in f:  # 한 줄씩 읽음
            count += 1
    return count

# 청크 단위 읽기 (바이너리)
def process_large_file(filepath: str, chunk_size: int = 8192):
    total_bytes = 0
    with open(filepath, "rb") as f:
        while chunk := f.read(chunk_size):
            total_bytes += len(chunk)
    return total_bytes

# 제너레이터로 CSV 처리
def read_csv_lazy(filepath: str):
    with open(filepath, "r", encoding="utf-8") as f:
        header = next(f).strip().split(",")
        for line in f:
            values = line.strip().split(",")
            yield dict(zip(header, values))

# 메모리 효율적 변환
def transform_file(src: str, dst: str):
    with open(src, "r") as fin, open(dst, "w") as fout:
        for line in fin:
            fout.write(line.upper())

# StringIO: 메모리 내 파일 객체
buffer = io.StringIO()
buffer.write("Hello\nWorld\n")
buffer.seek(0)
for line in buffer:
    print(f"  메모리 파일: {line.strip()}")

print("대용량 파일 처리 패턴 준비 완료")

Python code

import csv
import io

# CSV 쓰기
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["이름", "나이", "도시"])
writer.writerows([
    ["Alice", 30, "서울"],
    ["Bob", 25, "부산"],
    ["Carol", 35, "대전"],
])
csv_content = output.getvalue()
print(csv_content)

# CSV 읽기
reader = csv.DictReader(io.StringIO(csv_content))
for row in reader:
    print(f"  {row['이름']} ({row['나이']}) - {row['도시']}")

# CSV 방언(dialect) 설정
csv.register_dialect("custom",
    delimiter="|", quoting=csv.QUOTE_MINIMAL)

output2 = io.StringIO()
writer2 = csv.writer(output2, dialect="custom")
writer2.writerow(["a", "b", "c"])
print(f"커스텀: {output2.getvalue()}")

# 대용량 CSV 처리 (제너레이터)
def process_csv(file_obj):
    reader = csv.DictReader(file_obj)
    for row in reader:
        yield {k: v.strip() for k, v in row.items()}

# openpyxl 패턴 (설치 필요)
# from openpyxl import Workbook
# wb = Workbook()
# ws = wb.active
# ws.append(["이름", "점수"])
# ws.append(["Alice", 95])
# wb.save("output.xlsx")
print("CSV 처리 완료")

Python code

import json
from datetime import datetime, date
from dataclasses import dataclass, asdict

# 기본 직렬화/역직렬화
data = {"name": "Alice", "scores": [90, 85, 92], "active": True}
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(json_str)

parsed = json.loads(json_str)
print(parsed["name"])

# 커스텀 인코더
class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, (datetime, date)):
            return obj.isoformat()
        if isinstance(obj, set):
            return list(obj)
        return super().default(obj)

data = {
    "created": datetime.now(),
    "tags": {"python", "json"},
}
print(json.dumps(data, cls=CustomEncoder, ensure_ascii=False))

# dataclass → JSON
@dataclass
class User:
    name: str
    age: int
    email: str

user = User("Alice", 30, "alice@test.com")
print(json.dumps(asdict(user), ensure_ascii=False))

# JSON 파일 읽기/쓰기
# with open("data.json", "w", encoding="utf-8") as f:
#     json.dump(data, f, ensure_ascii=False, indent=2)
# with open("data.json", "r", encoding="utf-8") as f:
#     loaded = json.load(f)

Python code

import sqlite3

# 메모리 DB (테스트/데모용)
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row  # dict-like 접근

cursor = conn.cursor()

# 테이블 생성
cursor.execute("""
    CREATE TABLE users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER
    )
""")

# 데이터 삽입 (파라미터 바인딩)
users = [
    ("Alice", "alice@test.com", 30),
    ("Bob", "bob@test.com", 25),
    ("Carol", "carol@test.com", 35),
]
cursor.executemany(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    users
)
conn.commit()

# 조회
for row in cursor.execute("SELECT * FROM users WHERE age >= ?", (28,)):
    print(f"  {row['name']} ({row['age']}세) - {row['email']}")

# 집계
cursor.execute("SELECT AVG(age) as avg_age, COUNT(*) as cnt FROM users")
stats = cursor.fetchone()
print(f"평균 나이: {stats['avg_age']:.1f}, 총 {stats['cnt']}명")

# 컨텍스트 매니저로 안전하게
with conn:
    conn.execute("UPDATE users SET age = ? WHERE name = ?", (31, "Alice"))

conn.close()

Python code

# requests 라이브러리 패턴 시뮬레이션
from urllib.request import urlopen, Request
from urllib.error import URLError
import json

# 기본 GET 요청
def http_get(url: str, headers: dict = None) -> dict:
    req = Request(url, headers=headers or {})
    try:
        with urlopen(req, timeout=10) as resp:
            return {
                "status": resp.status,
                "body": resp.read().decode("utf-8"),
                "headers": dict(resp.headers),
            }
    except URLError as e:
        return {"error": str(e)}

# requests 스타일 API 시뮬레이션
class Response:
    def __init__(self, status=200, data=None):
        self.status_code = status
        self._data = data or {}

    def json(self):
        return self._data

    def raise_for_status(self):
        if self.status_code >= 400:
            raise Exception(f"HTTP {self.status_code}")

# 실사용 패턴 (requests 설치 필요)
# import requests
# resp = requests.get("https://api.example.com/users",
#     params={"page": 1},
#     headers={"Authorization": "Bearer TOKEN"},
#     timeout=10,
# )
# resp.raise_for_status()
# users = resp.json()

# httpx (비동기 지원)
# import httpx
# async with httpx.AsyncClient() as client:
#     resp = await client.get("https://api.example.com")

resp = Response(200, {"users": [{"name": "Alice"}]})
print(f"상태: {resp.status_code}")
print(f"데이터: {resp.json()}")

Python code

# BeautifulSoup 패턴 시뮬레이션
html = """
<html>
<body>
  <h1 class="title">Python 뉴스</h1>
  <ul id="articles">
    <li class="article">
      <a href="/post/1">첫 번째 글</a>
      <span class="date">2024-01-15</span>
    </li>
    <li class="article">
      <a href="/post/2">두 번째 글</a>
      <span class="date">2024-01-16</span>
    </li>
  </ul>
</body>
</html>
"""

# 간단한 파서 (실제는 BeautifulSoup 사용)
import re

# 태그 내용 추출
titles = re.findall(r"<a[^>]*>(.*?)</a>", html)
dates = re.findall(r'class="date">(.*?)</span>', html)
links = re.findall(r'href="([^"]*)"', html)

for title, date, link in zip(titles, dates, links):
    print(f"  [{date}] {title} ({link})")

# BeautifulSoup 실사용 패턴
# from bs4 import BeautifulSoup
# soup = BeautifulSoup(html, "html.parser")
# for article in soup.select("li.article"):
#     title = article.select_one("a").text
#     link = article.select_one("a")["href"]
#     date = article.select_one(".date").text
#     print(f"[{date}] {title} ({link})")

print("웹 스크래핑 패턴 시연 완료")

Python code

import os
import time
from pathlib import Path

# 간단한 폴링 기반 파일 감시
class SimpleWatcher:
    def __init__(self, path: str):
        self.path = Path(path)
        self._snapshot: dict[str, float] = {}

    def take_snapshot(self) -> dict[str, float]:
        snapshot = {}
        if self.path.is_dir():
            for f in self.path.iterdir():
                if f.is_file():
                    snapshot[str(f)] = f.stat().st_mtime
        return snapshot

    def check_changes(self) -> dict:
        new_snapshot = self.take_snapshot()
        changes = {"created": [], "modified": [], "deleted": []}

        for path, mtime in new_snapshot.items():
            if path not in self._snapshot:
                changes["created"].append(path)
            elif mtime != self._snapshot[path]:
                changes["modified"].append(path)

        for path in self._snapshot:
            if path not in new_snapshot:
                changes["deleted"].append(path)

        self._snapshot = new_snapshot
        return changes

# watchdog 라이브러리 패턴
# from watchdog.observers import Observer
# from watchdog.events import FileSystemEventHandler
#
# class MyHandler(FileSystemEventHandler):
#     def on_modified(self, event):
#         print(f"변경: {event.src_path}")
#     def on_created(self, event):
#         print(f"생성: {event.src_path}")
#
# observer = Observer()
# observer.schedule(MyHandler(), path=".", recursive=True)
# observer.start()

watcher = SimpleWatcher("/tmp")
snapshot = watcher.take_snapshot()
print(f"감시 중: {len(snapshot)}개 파일")
print("파일 감시 시스템 준비 완료")

Python code

import zipfile
import gzip
import io

# ZIP 파일 생성 (메모리 내)
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as zf:
    zf.writestr("hello.txt", "안녕하세요!")
    zf.writestr("data/config.json", '{"key": "value"}')

print(f"ZIP 크기: {buffer.tell()} bytes")

# ZIP 파일 읽기
buffer.seek(0)
with zipfile.ZipFile(buffer, "r") as zf:
    print(f"파일 목록: {zf.namelist()}")
    for name in zf.namelist():
        info = zf.getinfo(name)
        content = zf.read(name).decode("utf-8")
        print(f"  {name} ({info.compress_size}b): {content[:30]}")

# GZIP 압축
original = "Hello, World! " * 100
compressed = gzip.compress(original.encode("utf-8"))
decompressed = gzip.decompress(compressed).decode("utf-8")

print(f"\n원본: {len(original)} bytes")
print(f"압축: {len(compressed)} bytes")
print(f"비율: {len(compressed)/len(original)*100:.1f}%")
print(f"복원 일치: {original == decompressed}")

# tarfile 패턴
# import tarfile
# with tarfile.open("archive.tar.gz", "w:gz") as tar:
#     tar.add("src/", arcname="source")

Python code

import tempfile
import os

# 임시 파일 (자동 삭제)
with tempfile.NamedTemporaryFile(
    mode="w", suffix=".txt", prefix="myapp_",
    delete=True, encoding="utf-8"
) as tmp:
    tmp.write("임시 데이터")
    tmp.flush()
    print(f"임시 파일: {tmp.name}")
    print(f"존재: {os.path.exists(tmp.name)}")
# 블록 종료 후 자동 삭제

# 임시 디렉터리 (자동 삭제)
with tempfile.TemporaryDirectory(prefix="myapp_") as tmpdir:
    print(f"임시 디렉터리: {tmpdir}")
    # 디렉터리 안에 파일 생성
    filepath = os.path.join(tmpdir, "data.txt")
    with open(filepath, "w") as f:
        f.write("테스트 데이터")
    print(f"파일 존재: {os.path.exists(filepath)}")
# 블록 종료 후 디렉터리와 내용물 삭제

# SpooledTemporaryFile: 크기 초과 시 디스크로
with tempfile.SpooledTemporaryFile(
    max_size=1024, mode="w+", encoding="utf-8"
) as tmp:
    tmp.write("작은 데이터 → 메모리에 유지")
    tmp.seek(0)
    print(f"내용: {tmp.read()}")

# 안전한 임시 경로 생성
print(f"임시 디렉터리 경로: {tempfile.gettempdir()}")
print(f"고유 파일명: {tempfile.mktemp(suffix='.dat')}")