diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py
new file mode 100644
index 0000000..fb1b378
--- /dev/null
+++ b/backend/app/services/importer.py
@@ -0,0 +1,248 @@
+"""Single-file import pipeline.
+
+The Importer is the synchronous core of FC-2a's import path. A Celery task
+(tasks/import_file.py) instantiates one per job and calls import_one().
+
+The service is intentionally not async — Celery tasks are run in a
+synchronous worker process, and the DB session is a sync session passed in
+by the task. The Quart side uses an async session via the same models.
+"""
+
+import hashlib
+import shutil
+from dataclasses import dataclass
+from enum import Enum
+from pathlib import Path
+
+from PIL import Image
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from ..models import Artist, ImageRecord, ImportSettings, Tag, TagKind
+from ..models.tag import image_tag
+from ..utils.paths import derive_subdir, derive_top_level_artist, hash_suffixed_name
+from ..utils.slug import slugify
+from .thumbnailer import Thumbnailer
+
+
+class SkipReason(str, Enum):
+    too_small = "too_small"
+    too_transparent = "too_transparent"
+    single_color = "single_color"
+    duplicate_hash = "duplicate_hash"
+    invalid_image = "invalid_image"
+
+
+@dataclass(frozen=True)
+class ImportResult:
+    status: str  # 'imported' | 'skipped' | 'failed'
+    image_id: int | None = None
+    skip_reason: SkipReason | None = None
+    error: str | None = None
+
+
+IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tif", ".tiff"}
+VIDEO_EXTS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v", ".wmv", ".flv"}
+ALL_EXTS = IMAGE_EXTS | VIDEO_EXTS
+
+
+def is_supported(path: Path) -> bool:
+    return path.suffix.lower() in ALL_EXTS
+
+
+def is_video(path: Path) -> bool:
+    return path.suffix.lower() in VIDEO_EXTS
+
+
+def _mime_for(path: Path) -> str:
+    suffix = path.suffix.lower()
+    image_mimes = {
+        ".png": "image/png",
+        ".jpg": "image/jpeg",
+        ".jpeg": "image/jpeg",
+        ".gif": "image/gif",
+        ".webp": "image/webp",
+        ".bmp": "image/bmp",
+        ".tif": "image/tiff",
+        ".tiff": "image/tiff",
+    }
+    video_mimes = {
+        ".mp4": "video/mp4",
+        ".mov": "video/quicktime",
+        ".avi": "video/x-msvideo",
+        ".mkv": "video/x-matroska",
+        ".webm": "video/webm",
+        ".m4v": "video/x-m4v",
+        ".wmv": "video/x-ms-wmv",
+        ".flv": "video/x-flv",
+    }
+    return image_mimes.get(suffix) or video_mimes.get(suffix) or "application/octet-stream"
+
+
+def _sha256_of(path: Path, chunk_size: int = 1 << 20) -> str:
+    h = hashlib.sha256()
+    with path.open("rb") as fp:
+        while True:
+            block = fp.read(chunk_size)
+            if not block:
+                break
+            h.update(block)
+    return h.hexdigest()
+
+
+class Importer:
+    """Imports one file at a time. Constructed per-task by the Celery task."""
+
+    def __init__(
+        self,
+        session: Session,
+        images_root: Path,
+        import_root: Path,
+        thumbnailer: Thumbnailer,
+        settings: ImportSettings,
+    ):
+        self.session = session
+        self.images_root = images_root
+        self.import_root = import_root
+        self.thumbnailer = thumbnailer
+        self.settings = settings
+
+    def import_one(self, source: Path) -> ImportResult:
+        """The full pipeline for a single file.
+
+        Returns an ImportResult; callers (the Celery task) are responsible for
+        translating that into ImportTask.status and ImageRecord linkage.
+        """
+        if not is_supported(source):
+            return ImportResult(
+                status="skipped", skip_reason=SkipReason.invalid_image,
+                error=f"unsupported extension {source.suffix}",
+            )
+
+        # Compute file dimensions (images only) and apply filters.
+        width = height = None
+        has_alpha = False
+        if not is_video(source):
+            try:
+                with Image.open(source) as im:
+                    im.verify()
+                with Image.open(source) as im:
+                    width, height = im.size
+                    has_alpha = im.mode in ("RGBA", "LA") or (
+                        im.mode == "P" and "transparency" in im.info
+                    )
+            except Exception as exc:
+                return ImportResult(
+                    status="skipped", skip_reason=SkipReason.invalid_image, error=str(exc),
+                )
+
+            if (self.settings.min_width and width < self.settings.min_width) or (
+                self.settings.min_height and height < self.settings.min_height
+            ):
+                return ImportResult(
+                    status="skipped", skip_reason=SkipReason.too_small,
+                    error=f"{width}x{height} below min {self.settings.min_width}x{self.settings.min_height}",
+                )
+
+            if self.settings.skip_transparent and has_alpha:
+                pct = self._transparency_pct(source)
+                if pct >= self.settings.transparency_threshold:
+                    return ImportResult(
+                        status="skipped", skip_reason=SkipReason.too_transparent,
+                        error=f"{pct:.2%} transparent",
+                    )
+
+        # Hash dedup.
+        sha = _sha256_of(source)
+        existing_stmt = select(ImageRecord).where(ImageRecord.sha256 == sha)
+        existing = self.session.execute(existing_stmt).scalar_one_or_none()
+        if existing:
+            return ImportResult(
+                status="skipped", skip_reason=SkipReason.duplicate_hash,
+                image_id=existing.id, error="sha256 already present",
+            )
+
+        # Destination path.
+        subdir = derive_subdir(source, self.import_root)
+        dest_dir = self.images_root / subdir if subdir else self.images_root
+        dest_dir.mkdir(parents=True, exist_ok=True)
+        dest_name = hash_suffixed_name(source.stem, sha, source.suffix)
+        dest = dest_dir / dest_name
+
+        # Atomic copy: write to .partial, rename. Avoids half-imported files on crash.
+        partial = dest.with_suffix(dest.suffix + ".partial")
+        shutil.copy2(source, partial)
+        partial.rename(dest)
+
+        record = ImageRecord(
+            path=str(dest),
+            sha256=sha,
+            size_bytes=dest.stat().st_size,
+            mime=_mime_for(source),
+            width=width,
+            height=height,
+            origin="imported_filesystem",
+            integrity_status="unknown",
+        )
+        self.session.add(record)
+        self.session.flush()
+
+        # Folder→artist auto-tag.
+        artist_name = derive_top_level_artist(source, self.import_root)
+        if artist_name:
+            self._attach_artist_tag(record.id, artist_name)
+
+        # Thumbnail is queued separately by the calling task; the importer
+        # does not generate thumbnails inline so the import queue stays moving.
+
+        self.session.commit()
+        return ImportResult(status="imported", image_id=record.id)
+
+    def _attach_artist_tag(self, image_id: int, artist_name: str) -> None:
+        """Idempotently creates Artist + artist tag, attaches to image."""
+        slug = slugify(artist_name)
+        artist = self.session.execute(
+            select(Artist).where(Artist.slug == slug)
+        ).scalar_one_or_none()
+        if artist is None:
+            artist = Artist(name=artist_name, slug=slug, is_subscription=False)
+            self.session.add(artist)
+            self.session.flush()
+
+        tag = self.session.execute(
+            select(Tag).where(Tag.name == artist_name).where(Tag.kind == TagKind.artist)
+        ).scalar_one_or_none()
+        if tag is None:
+            tag = Tag(name=artist_name, kind=TagKind.artist)
+            self.session.add(tag)
+            self.session.flush()
+
+        # Idempotent association; using a raw insert + on-conflict-do-nothing
+        # is cleaner here than fetching first, but for a sync session we just
+        # check existence.
+        already = self.session.execute(
+            select(image_tag.c.tag_id)
+            .where(image_tag.c.image_record_id == image_id)
+            .where(image_tag.c.tag_id == tag.id)
+        ).scalar_one_or_none()
+        if already is None:
+            self.session.execute(
+                image_tag.insert().values(
+                    image_record_id=image_id, tag_id=tag.id, source="auto",
+                )
+            )
+
+    def _transparency_pct(self, source: Path) -> float:
+        """Fraction of fully-transparent pixels in the image. 0.0 if no alpha."""
+        with Image.open(source) as im:
+            if im.mode not in ("RGBA", "LA") and not (
+                im.mode == "P" and "transparency" in im.info
+            ):
+                return 0.0
+            if im.mode != "RGBA":
+                im = im.convert("RGBA")
+            alpha = im.getchannel("A")
+            histogram = alpha.histogram()
+            transparent = histogram[0]
+            total = sum(histogram)
+            return transparent / total if total else 0.0
diff --git a/tests/test_importer.py b/tests/test_importer.py
new file mode 100644
index 0000000..5739355
--- /dev/null
+++ b/tests/test_importer.py
@@ -0,0 +1,132 @@
+"""Importer integration tests.
+
+These exercise the real Importer against a real Postgres + a real filesystem
+(via tmp_path). The DB fixture rolls back after each test so the import_root
+fixture is fresh per-test.
+"""
+
+from pathlib import Path
+
+import pytest
+from PIL import Image
+from sqlalchemy import select
+
+from backend.app.models import Artist, ImageRecord, ImportSettings, Tag, TagKind
+from backend.app.services.importer import Importer, SkipReason
+from backend.app.services.thumbnailer import Thumbnailer
+
+
+@pytest.fixture
+def import_layout(tmp_path):
+    import_root = tmp_path / "import"
+    images_root = tmp_path / "images"
+    import_root.mkdir()
+    images_root.mkdir()
+    return import_root, images_root
+
+
+def _make_jpeg(path: Path, size: tuple[int, int] = (800, 600)):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGB", size, color=(120, 200, 80)).save(path, "JPEG")
+
+
+def _make_png_rgba(path: Path, size: tuple[int, int], alpha: int):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGBA", size, color=(120, 200, 80, alpha)).save(path, "PNG")
+
+
+@pytest.fixture
+def importer(db_sync, import_layout):
+    import_root, images_root = import_layout
+    settings = db_sync.execute(select(ImportSettings).where(ImportSettings.id == 1)).scalar_one()
+    thumbnailer = Thumbnailer(images_root=images_root)
+    return Importer(
+        session=db_sync,
+        images_root=images_root,
+        import_root=import_root,
+        thumbnailer=thumbnailer,
+        settings=settings,
+    )
+
+
+def test_import_one_happy_path(importer, import_layout):
+    import_root, images_root = import_layout
+    src = import_root / "Alice" / "first.jpg"
+    _make_jpeg(src)
+    result = importer.import_one(src)
+    assert result.status == "imported"
+    record = importer.session.get(ImageRecord, result.image_id)
+    assert record.path.startswith(str(images_root / "Alice"))
+    assert record.path.endswith(".jpg")
+
+
+def test_folder_creates_artist_and_tag(importer, import_layout):
+    import_root, _ = import_layout
+    src = import_root / "Alice" / "first.jpg"
+    _make_jpeg(src)
+    importer.import_one(src)
+
+    artist = importer.session.execute(
+        select(Artist).where(Artist.slug == "alice")
+    ).scalar_one()
+    assert artist.name == "Alice"
+
+    tag = importer.session.execute(
+        select(Tag).where(Tag.name == "Alice").where(Tag.kind == TagKind.artist)
+    ).scalar_one()
+    assert tag.id is not None
+
+
+def test_dedup_by_hash(importer, import_layout):
+    import_root, _ = import_layout
+    src1 = import_root / "Alice" / "first.jpg"
+    src2 = import_root / "Bob" / "duplicate.jpg"
+    _make_jpeg(src1)
+    src2.parent.mkdir(parents=True, exist_ok=True)
+    src2.write_bytes(src1.read_bytes())
+
+    r1 = importer.import_one(src1)
+    r2 = importer.import_one(src2)
+    assert r1.status == "imported"
+    assert r2.status == "skipped"
+    assert r2.skip_reason == SkipReason.duplicate_hash
+
+
+def test_min_width_filter(importer, import_layout):
+    import_root, _ = import_layout
+    importer.settings.min_width = 1000  # mutate the in-memory copy for this test
+    src = import_root / "small.jpg"
+    _make_jpeg(src, size=(500, 500))
+    result = importer.import_one(src)
+    assert result.status == "skipped"
+    assert result.skip_reason == SkipReason.too_small
+
+
+def test_transparent_filter(importer, import_layout):
+    import_root, _ = import_layout
+    importer.settings.skip_transparent = True
+    importer.settings.transparency_threshold = 0.5
+    src = import_root / "ghost.png"
+    _make_png_rgba(src, size=(100, 100), alpha=0)  # fully transparent
+    result = importer.import_one(src)
+    assert result.status == "skipped"
+    assert result.skip_reason == SkipReason.too_transparent
+
+
+def test_unsupported_extension(importer, import_layout):
+    import_root, _ = import_layout
+    src = import_root / "Alice" / "notes.txt"
+    src.parent.mkdir(parents=True, exist_ok=True)
+    src.write_text("hello")
+    result = importer.import_one(src)
+    assert result.status == "skipped"
+    assert result.skip_reason == SkipReason.invalid_image
+
+
+def test_root_level_file_has_no_artist(importer, import_layout):
+    import_root, _ = import_layout
+    src = import_root / "loose.jpg"
+    _make_jpeg(src)
+    importer.import_one(src)
+    artists = importer.session.execute(select(Artist)).scalars().all()
+    assert artists == []