File Uploads 심화

심화 (Deep Dive)예상 1-2시간

파일 업로드의 고급 기능과 보안 전략을 다룹니다.

학습 목표#

• 업로드 설정 상세
• 진행률 추적
• 다중 파일 업로드
• 보안 및 검증
• 저장소 통합

업로드 설정#

allow_upload 옵션#

업로드 필드는 joined()에서 allow_upload()로 연다. 첫 인자가 필드 이름이고 나머지는 키워드다.

async def joined(self):
    self.allow_upload(
        "avatar",                    # 업로드 필드 식별자
        accept=[".jpg", ".png"],     # 허용 확장자
        max_entries=1,               # 동시에 진행할 수 있는 업로드 수
        max_file_size=5*1024*1024,   # 5MB
        chunk_size=64*1024,          # 64KB 청크
        auto_upload=True,            # 선택 즉시 업로드
    )

max_entries는 진행 중인 업로드의 수다. consume_uploads()로 소비한 업로드는 세지 않으므로, 한 개짜리 필드로도 파일을 하나씩 계속 올릴 수 있다.

다양한 설정 예#

async def joined(self):
    # 프로필 이미지 (단일, 작은 파일)
    self.allow_upload(
        "avatar",
        accept=[".jpg", ".jpeg", ".png", ".gif", ".webp"],
        max_file_size=2 * 1024 * 1024,  # 2MB
    )

    # 문서 업로드 (다중, 큰 파일)
    self.allow_upload(
        "documents",
        accept=[".pdf", ".doc", ".docx", ".xls", ".xlsx"],
        max_entries=10,
        max_file_size=50 * 1024 * 1024,  # 50MB
        chunk_size=256 * 1024,  # 256KB 청크
    )

    # 갤러리 이미지 (다중)
    self.allow_upload(
        "photos",
        accept=[".jpg", ".jpeg", ".png"],
        max_entries=20,
        max_file_size=10 * 1024 * 1024,  # 10MB
        auto_upload=False,  # 수동 업로드
    )

컴포넌트 구현#

기본 구조#

from wireview import Component


class XFileUploader(Component):
    class Meta:
        template_name = 'uploader/upload_form.html'

    async def joined(self):
        self.allow_upload(
            "files",
            accept=[".pdf", ".jpg", ".png"],
            max_entries=5,
            max_file_size=10 * 1024 * 1024,
        )

    async def upload_files(self):
        """업로드된 파일 처리"""
        async for upload in self.consume_uploads("files"):
            # upload는 ConsumedUpload 인스턴스
            path = await upload.save_to("uploads/")
            print(f"Saved: {path}")

    async def cancel_file(self, ref: str):
        """목록의 × 버튼. 자세한 것은 아래 "업로드 취소"에서 다룬다"""
        await self.cancel_upload("files", ref)

템플릿#

{% load wireview %}
<div {% tag_header %} class="uploader">
  <div class="upload-zone" {% upload_drop_zone "files" %}>
    {% upload_input "files" %}
    <p>Drop files here or click to browse</p>
  </div>

  <!-- 업로드 목록 -->
  <ul class="upload-list">
    {% for entry in this.uploads.files %}
      <li class="upload-entry {{ entry.status }}">
        <span class="name">{{ entry.client_name }}</span>
        <span class="size">{{ entry.client_size|filesizeformat }}</span>

        {% if entry.status == 'uploading' %}
          <div class="progress">
            <div class="bar" style="width: {{ entry.progress }}%"></div>
          </div>
        {% elif entry.status == 'completed' %}
          <span class="status">✓ Ready</span>
        {% elif entry.status == 'error' %}
          <span class="error">{{ entry.errors|join:", " }}</span>
        {% endif %}

        <button {% on "click" "cancel_file" ref=entry.ref %}>×</button>
      </li>
    {% endfor %}
  </ul>

  <button {% on "click" "upload_files" %}>Save Files</button>
</div>

진행률 추적#

UploadEntry 속성#

상태별 처리#

{% if entry.status == 'pending' %}
  <span>Waiting...</span>

{% elif entry.status == 'uploading' %}
  <div class="progress">{{ entry.progress }}%</div>

{% elif entry.status == 'completed' %}
  <span class="success">Upload complete</span>

{% elif entry.status == 'error' %}
  <span class="error">
    {% for error in entry.errors %}
      {{ error }}<br>
    {% endfor %}
  </span>

{% elif entry.status == 'cancelled' %}
  <span>Cancelled</span>
{% endif %}

파일 처리#

ConsumedUpload API#

async def process_upload(self):
    async for upload in self.consume_uploads("files"):
        # 파일 정보
        print(upload.name)         # 원본 파일명
        print(upload.size)         # 크기 (bytes)
        print(upload.content_type) # MIME 타입
        print(upload.ref)          # 참조 ID

        # 내용 읽기. 임시 파일을 읽을 뿐이라 몇 번이든 된다
        content = upload.read()    # bytes로 읽기
        with upload.open("rb") as f:
            header = f.read(16)

        # 저장. 한 업로드에 한 번만 — save_to가 임시 파일을 지우고 업로드를 소비한다
        path = await upload.save_to("uploads/")

원래 이름 대신 다른 이름으로 저장하려면 filename=을 넘긴다. 같은 업로드를 두 번 save_to()하면 두 번째는 RuntimeError("Upload already consumed")다 — 두 곳에 저장하려면 read()한 바이트를 직접 쓴다.

path = await upload.save_to("avatars/", filename=f"user_{self.user.pk}.jpg")

save_to()를 부르지 않아도 루프가 다음 업로드로 넘어가면 그 업로드는 소비되고 임시 파일은 지워진다.

Django Storage 통합#

save_to()는 Django의 default storage를 사용합니다:

# settings.py
STORAGES = {
    "default": {"BACKEND": "storages.backends.s3boto3.S3Boto3Storage"},
    "staticfiles": {"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"},
}

# 컴포넌트
async def save_to_s3(self):
    async for upload in self.consume_uploads("files"):
        # S3에 저장됨
        path = await upload.save_to("media/uploads/")

직접 처리#

from PIL import Image

async def process_image(self):
    async for upload in self.consume_uploads("avatar"):
        # Pillow로 이미지 처리
        with upload.open("rb") as f:
            img = Image.open(f)
            img = img.resize((200, 200))

            # 처리된 이미지 저장
            from io import BytesIO
            buffer = BytesIO()
            img.save(buffer, format='JPEG')
            # ...

보안#

확장자 검증#

self.allow_upload("images", accept=[".jpg", ".png"])  # 허용된 확장자만

Magic Bytes 검증#

Wireview는 파일 시그니처를 자동 검증합니다:

# wireview/features/uploads.py
MAGIC_BYTES = {
    ".jpg": [b"\xff\xd8\xff"],
    ".png": [b"\x89PNG\r\n\x1a\n"],
    ".pdf": [b"%PDF"],
    # ...
}

.jpg 확장자지만 실제로 PNG인 경우 → 에러

크기 제한#

self.allow_upload("images", max_file_size=5 * 1024 * 1024)  # 5MB

수동 검증#

errors: list[str] = []

async def save_files(self):
    async for upload in self.consume_uploads("files"):
        # 추가 검증. 건너뛴 업로드도 소비된 것이라 임시 파일은 지워진다
        if upload.size > 1024 * 1024:
            self.errors = [*self.errors, f"{upload.name}: too large"]
            continue

        if not self._is_safe_filename(upload.name):
            self.errors = [*self.errors, f"{upload.name}: invalid filename"]
            continue

        await upload.save_to("uploads/")

def _is_safe_filename(self, name: str) -> bool:
    import re
    # 안전한 문자만 허용
    return bool(re.match(r'^[\w\-. ]+$', name))

소유권과 수명#

업로드는 연결(connection)의 것이지 컴포넌트 id의 것이 아니다. 컴포넌트 id는 페이지 안에서만 고유하고, 템플릿이 {% component 'X' id="bookmarks" %}처럼 id를 고정하면 같은 페이지를 연 두 연결이 같은 id를 쓴다. 그래서 서버가 업로드에 쓰는 모든 것에 연결 id가 붙는다 (#77).

연결 id는 WebSocket 연결이 열려 세션이 시작될 때 secrets.token_urlsafe(16)으로 발급된다. 채널 레이어 주소인 channel_name과 달리 URL에 넣어도 되는 값이다.

엔드포인트는 서버가 config 업로드 op으로 클라이언트에 보내 준다. 경로를 직접 적어 두지 않았다면 앱에서 고칠 것은 없다.

정리 시점#

브라우저도 같은 때 그 컴포넌트의 업로드 상태(기다리는 파일, 진행 중인 요청, 미리보기)를 버린다. 상세는 chunked-uploads.md의 「브라우저 쪽 수명」.

경로와 토큰 양쪽에 소유자가 들어가므로 한 연결의 leave가 다른 연결의 파일을 건드릴 수 없다. A에서 받은 토큰을 B의 URL로 보내면 서명은 검증되지만 소유자가 URL과 달라 403이다.

워커가 여럿일 때#

청크 HTTP 요청은 아무 워커에나 닿아도 된다(#83). 엔드포인트는 업로드 상태를 하나도 들고 있지 않고, 서명된 토큰만으로 판단해 토큰에서 계산한 경로에 쓴다. 대신 워커들이 청크 저장소(WIREVIEW["UPLOAD_TEMP_DIR"], None이면 시스템 임시 디렉터리 — 한 호스트라면 이미 공유다)와 서명 키를 공유해야 한다.

• 설정된 경로는 없으면 만들어지고, 쓸 수 없으면 ImproperlyConfigured로 실패한다 — 공유 볼륨을 지정했는데 말없이 로컬 디스크를 쓰는 일은 없다. manage.py check의 wireview.W008이 미리 잡는다.
• 정상 disconnect 없이 프로세스가 죽으면 청크 파일이 남는다. 남는 양은 UPLOAD_TOKEN_MAX_AGE로 묶이고, 쓰기 경로가 프로세스당 10분에 한 번 청소한다. cron으로 돌리려면 manage.py wireview_upload_gc.
• 서버 쪽 동작 전체는 docs/features/chunked-uploads.md, 배포 조건은 docs/DEPLOYMENT.md.

다중 업로드 필드#

여러 업로드 설정#

async def joined(self):
    self.allow_upload("avatar", max_entries=1)
    self.allow_upload("documents", max_entries=10)
    self.allow_upload("gallery", max_entries=20)

필드별 처리#

async def save_all(self):
    # 아바타
    async for upload in self.consume_uploads("avatar"):
        await upload.save_to("avatars/")

    # 문서
    async for upload in self.consume_uploads("documents"):
        await upload.save_to("documents/")

    # 갤러리
    async for upload in self.consume_uploads("gallery"):
        await upload.save_to("gallery/")

업로드 취소#

사용자 취소#

async def cancel_file(self, ref: str):
    """특정 파일 업로드 취소"""
    await self.cancel_upload("files", ref)

템플릿#

<button {% on "click" "cancel_file" ref=entry.ref %}>Cancel</button>

드래그 앤 드롭#

업로드 태그는 넷이다. upload_input과 upload_preview는 요소를 그리고, upload_drop_zone과 upload_button은 사용자가 쓴 요소에 붙일 속성을 낸다.

{% upload_drop_zone "필드" %}를 단 요소에 파일을 떨어뜨리면 그 필드로 업로드된다. 파일을 끌고 들어오는 동안 요소에 wireview-drag-over 클래스가 붙는다.

<div class="upload-zone" {% upload_drop_zone "files" %}>
  여기에 파일을 놓으세요
  {% upload_input "files" %}
</div>

.upload-zone {
  border: 2px dashed #ccc;
  padding: 2rem;
  text-align: center;
}

.upload-zone.wireview-drag-over {
  border-color: #007bff;
  background: #f0f7ff;
}

이미지 미리보기#

{% upload_preview %} 태그로 업로드 전 이미지 미리보기를 표시합니다.

기본 사용법#

{% load wireview %}

<div {% tag_header %}>
  <h3>프로필 사진 업로드</h3>

  {% upload_input "avatar" %}

  <!-- 업로드된 이미지 미리보기 -->
  {% for entry in this.uploads.avatar %}
    <div class="preview-container">
      {% upload_preview entry class="w-32 h-32 rounded-full object-cover" %}
      <p>{{ entry.client_name }}</p>
    </div>
  {% endfor %}
</div>

갤러리 미리보기#

<div class="gallery-upload">
  {% upload_input "photos" %}  {# max_entries가 1보다 크면 multiple이 자동으로 붙는다 #}

  <div class="grid grid-cols-4 gap-4">
    {% for entry in this.uploads.photos %}
      <div class="relative">
        {% upload_preview entry class="w-full h-32 object-cover rounded" %}

        <!-- 업로드 진행률 오버레이 -->
        {% if entry.status == 'uploading' %}
          <div class="absolute inset-0 bg-black/50 flex items-center justify-center">
            <span class="text-white">{{ entry.progress }}%</span>
          </div>
        {% endif %}

        <!-- 취소 버튼 -->
        <button
          {% on "click" "cancel_file" ref=entry.ref %}
          class="absolute top-1 right-1 bg-red-500 text-white rounded-full w-6 h-6"
        >×</button>
      </div>
    {% endfor %}
  </div>
</div>

미리보기가 표시되지 않는 경우#

• 비이미지 파일: PDF, DOC 등은 미리보기가 표시되지 않습니다
• 파일 미선택: 아직 파일을 선택하지 않은 경우
• 브라우저 호환성: 구형 브라우저에서는 Blob URL을 지원하지 않을 수 있습니다

외부 스토리지 업로드 (S3/GCS)#

대용량 파일은 Django 서버를 거치지 않고 직접 클라우드 스토리지로 업로드할 수 있습니다.

S3 직접 업로드#

콜백 이름은 _로 시작한다. _가 없는 메서드는 클라이언트가 부를 수 있는 이벤트 핸들러로 노출된다.

import boto3
from wireview import Component, ExternalUploadMeta


class XDocumentUploader(Component):
    class Meta:
        template_name = "documents/uploader.html"

    async def joined(self):
        self.allow_upload(
            "documents",
            accept=[".pdf", ".doc", ".docx"],
            max_file_size=100 * 1024 * 1024,  # 100MB
            external=self._presign_s3_upload,  # 외부 업로드 콜백
        )

    def _presign_s3_upload(self, entry, component):
        """S3 presigned URL 생성"""
        s3 = boto3.client("s3")

        key = f"documents/{entry.ref}/{entry.client_name}"
        url = s3.generate_presigned_url(
            "put_object",
            Params={
                "Bucket": "my-bucket",
                "Key": key,
                "ContentType": entry.client_type,
            },
            ExpiresIn=3600,
        )

        return ExternalUploadMeta(
            uploader="S3",
            url=url,
            headers={"Content-Type": entry.client_type},
        )

    async def save_document(self):
        # 업로드 완료 후 DB에 기록
        async for upload in self.consume_uploads("documents"):
            key = f"documents/{upload.ref}/{upload.name}"
            await Document.objects.acreate(
                name=upload.name,
                s3_key=key,
                size=upload.size,
            )

GCS 직접 업로드#

from google.cloud import storage
from wireview import Component, ExternalUploadMeta


class XImageUploader(Component):
    class Meta:
        template_name = "images/uploader.html"

    async def joined(self):
        self.allow_upload(
            "images",
            accept=[".jpg", ".png", ".webp"],
            external=self._presign_gcs_upload,
        )

    def _presign_gcs_upload(self, entry, component):
        """GCS signed URL 생성"""
        client = storage.Client()
        bucket = client.bucket("my-bucket")
        blob = bucket.blob(f"images/{entry.ref}/{entry.client_name}")

        url = blob.generate_signed_url(
            version="v4",
            expiration=3600,
            method="PUT",
            content_type=entry.client_type,
        )

        return ExternalUploadMeta(
            uploader="GCS",
            url=url,
            headers={"Content-Type": entry.client_type},
        )

CORS 설정#

외부 업로드를 위해 스토리지 버킷에 CORS 설정이 필요합니다:

S3 CORS:

{
  "CORSRules": [{
    "AllowedOrigins": ["https://your-domain.com"],
    "AllowedMethods": ["PUT"],
    "AllowedHeaders": ["*"],
    "MaxAgeSeconds": 3600
  }]
}

일반 업로드 vs 외부 업로드#

에러 처리#

공통 에러#

에러 표시#

{% if entry.errors %}
  <div class="errors">
    {% for error in entry.errors %}
      <p class="error">{{ error }}</p>
    {% endfor %}
  </div>
{% endif %}

완성된 예제#

class XProfileEditor(Component):
    class Meta:
        template_name = 'profile/editor.html'

    user_id: int
    avatar_url: str = ""

    async def joined(self):
        self.allow_upload(
            "avatar",
            accept=[".jpg", ".jpeg", ".png", ".gif"],
            max_file_size=2 * 1024 * 1024,
        )

    async def save_avatar(self):
        async for upload in self.consume_uploads("avatar"):
            try:
                path = await upload.save_to(
                    "avatars/",
                    filename=f"user_{self.user_id}.jpg"
                )
                self.avatar_url = str(path)

                # DB 업데이트
                await User.objects.filter(id=self.user_id).aupdate(
                    avatar=path
                )

            except ValueError as e:
                await self.put_flash("error", str(e))

다음 단계#