CFileDialog 구현 방법
PySide6(PyQt6)를 이용해서 나만의 QFileDialog를 만들고 싶어. 이름은 CFileDialog라고 하자. 그냥 Qt의 내장 다이얼로그와 비슷하면 돼 ## UI 설명 가장 위에는 현재 경로가 콤보 박스로 있고, 콤보박스를 누르면 하위 경로들을 선택할 수 있게 된다 예를 들어 지금 /a/b/c/d 에서 시작했으면 /a/b/c/d /a/b/c /a/b /a / 이런 식으로 나와야하고, 경로 클릭 시 그 경로로 다이얼로그 이동할 것. 그 옆에는 Back, Forward, Parent Directory, Create New Folder 버튼이 있다. 가운데 리스트뷰에는 현재 폴더의 하위 폴더, 파일들이 보인다. (모드에 따라 보이는게 다름, 폴더 선택 모드냐 파일 선택 모드냐) 아래에는 경로를 입력을 할 수 있는 LineEdit이 있고 경로 입력 시에 자동완성 될만한게 QCompleter로 나타난다. 이 때 네트워크 드라이브를 사용하므로 QCompleter가 매우 느릴 수 있는데, QCompleter의 역할을 쓰레드로 빼면 좋을 것 같음. 현재 경로에서 가장 일치하는 상위 10개 정도 경로만 Completer로 사용하고 경로가 완성되면 다시 리프레시 해서 상위 10개를 찾는 식으로 하면 좋을 것 같음. (중요) 그 아래에는 Type 선택 콤보박스가 있다. 그 외 사용법은 QFileDialog와 거의 동일하게 하면 좋겠음. 파일/폴더 얻을 지 Save용인지 Load용인지
좋아요. 이건 꽤 본격적인 커스텀 다이얼로그 설계예요.
PySide6(PyQt6 동일 구조)에서 Qt의 기본 QFileDialog를 재현하면서 더 유연하게 만들기 위해선 QDialog 기반의 클래스 로 직접 구현하는 게 가장 깔끔합니다.
아래는 전체적인 구조와 주요 포인트를 단계별로 설명할게요.
핵심 구조 개요
CFileDialog는 기본적으로 QDialog를 상속받습니다.
내부 구성요소는 다음과 같습니다.
-
상단 영역 :
-
경로 콤보박스 (
QComboBox) -
Back / Forward / Up / New Folder 버튼 (
QToolButton)
-
-
중앙 영역 :
-
파일/폴더 목록 (
QListVieworQTreeView) -
모델은
QFileSystemModel사용
-
-
하단 영역 :
-
경로 입력창 (
QLineEdit) + 자동완성 (QCompleter) -
Type 선택 콤보박스 (
QComboBox) -
확인/취소 버튼 (
QDialogButtonBox)
-
주요 동작 설계
1. 경로 콤보박스
현재 경로를 /a/b/c/d 형태로 표시하고, 클릭 시 아래처럼 보여야 함:
/a/b/c/d
/a/b/c
/a/b
/a
/
이건 QComboBox를 매번 경로 변경 시 갱신하는 방식으로 구현 가능.
def update_path_combo(self, path):
self.path_combo.clear()
parts = Path(path).parts
for i in range(len(parts), 0, -1):
subpath = os.path.join(*parts[:i])
self.path_combo.addItem(subpath)
콤보박스에서 항목 선택 시 QFileSystemModel의 root를 해당 경로로 변경.
2. 탐색 버튼들
-
Back / Forward : 간단히
deque를 이용한 히스토리 관리. -
Up (Parent) :
os.path.dirname(current_path)로 이동. -
New Folder :
QInputDialog.getText()로 폴더명 받아os.makedirs().
3. 파일/폴더 리스트
QFileSystemModel + QListView 또는 QTreeView 사용.
선택 모드에 따라 표시되는 항목 제한 가능:
if self.select_mode == 'folder':
self.model.setFilter(QDir.AllDirs | QDir.NoDotAndDotDot)
else:
self.model.setFilter(QDir.AllEntries | QDir.NoDotAndDotDot)
QListView.clicked 시 LineEdit 업데이트.
4. 경로 입력 + 자동완성
핵심 포인트는 QCompleter를 별도 쓰레드로 관리 하는 부분이에요.
대용량 네트워크 드라이브에서 디렉토리 자동완성은 느리므로,
다음과 같은 비동기 구조를 씁니다.
Worker Thread
class PathCompleterWorker(QObject):
paths_ready = Signal(list)
def __init__(self, base_path, prefix):
super().__init__()
self.base_path = base_path
self.prefix = prefix
def run(self):
matches = []
for root, dirs, files in os.walk(self.base_path):
for d in dirs:
if d.startswith(self.prefix):
matches.append(os.path.join(root, d))
if len(matches) >= 10:
break
if len(matches) >= 10:
break
self.paths_ready.emit(matches)
CFileDialog에서 입력창 내용이 바뀔 때마다 이 쓰레드 실행 → QCompleter 갱신.
5. Type 선택 콤보박스
일반 파일 다이얼로그처럼 *.txt, *.png 등의 필터를 표시:
self.type_combo.addItems(["All Files (*)", "Text Files (*.txt)", "Images (*.png *.jpg)"])
currentTextChanged 시 모델 필터 갱신.
6. QFileDialog처럼 동작하게 만들기
외부 인터페이스는 기존 QFileDialog와 유사하게:
dlg = CFileDialog(mode='open', select_mode='file')
if dlg.exec():
selected_path = dlg.selected_path()
옵션:
-
mode:"open"/"save" -
select_mode:"file"/"folder"
전체 구조 코드 예시 (요약 버전)
from PySide6.QtWidgets import *
from PySide6.QtCore import *
from PySide6.QtGui import *
import os
from pathlib import Path
from collections import deque
class CFileDialog(QDialog):
def __init__(self, start_path=None, mode='open', select_mode='file', parent=None):
super().__init__(parent)
self.mode = mode
self.select_mode = select_mode
self.history = deque()
self.forward_stack = deque()
self.current_path = start_path or os.getcwd()
self._build_ui()
self._connect_signals()
self._load_path(self.current_path)
def _build_ui(self):
self.setWindowTitle("CFileDialog")
# 상단
self.path_combo = QComboBox()
self.back_btn = QToolButton(text="←")
self.forward_btn = QToolButton(text="→")
self.up_btn = QToolButton(text="↑")
self.new_folder_btn = QToolButton(text="+")
top_layout = QHBoxLayout()
top_layout.addWidget(self.back_btn)
top_layout.addWidget(self.forward_btn)
top_layout.addWidget(self.up_btn)
top_layout.addWidget(self.new_folder_btn)
top_layout.addWidget(self.path_combo)
# 중앙
self.model = QFileSystemModel()
self.model.setRootPath("")
self.view = QListView()
self.view.setModel(self.model)
# 하단
self.path_edit = QLineEdit()
self.type_combo = QComboBox()
self.type_combo.addItems(["All Files (*)"])
self.button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
bottom_layout = QHBoxLayout()
bottom_layout.addWidget(self.path_edit)
bottom_layout.addWidget(self.type_combo)
layout = QVBoxLayout(self)
layout.addLayout(top_layout)
layout.addWidget(self.view)
layout.addLayout(bottom_layout)
layout.addWidget(self.button_box)
def _connect_signals(self):
self.path_combo.currentTextChanged.connect(self._load_path)
self.up_btn.clicked.connect(self._go_up)
self.back_btn.clicked.connect(self._go_back)
self.forward_btn.clicked.connect(self._go_forward)
self.new_folder_btn.clicked.connect(self._create_folder)
self.view.clicked.connect(self._item_clicked)
self.button_box.accepted.connect(self.accept)
self.button_box.rejected.connect(self.reject)
def _load_path(self, path):
if not os.path.exists(path): return
self.current_path = path
self.view.setRootIndex(self.model.index(path))
self.update_path_combo(path)
self.path_edit.setText(path)
def update_path_combo(self, path):
self.path_combo.blockSignals(True)
self.path_combo.clear()
parts = Path(path).parts
for i in range(len(parts), 0, -1):
self.path_combo.addItem(os.path.join(*parts[:i]))
self.path_combo.blockSignals(False)
def _item_clicked(self, index):
path = self.model.filePath(index)
if os.path.isdir(path):
self._load_path(path)
else:
self.path_edit.setText(path)
def selected_path(self):
return self.path_edit.text()
# 기타 버튼 함수 생략 (Back, Forward, Up, etc.)
이 설계대로라면 Qt 내장 QFileDialog와 거의 동일한 인터페이스를 가지되,
확장성과 사용자 지정성이 훨씬 높습니다.
다음 단계로는 Completer를 쓰레드로 실제 연결하는 부분 을 구현해보면 완성형에 가까워집니다.
원한다면 그 부분만 이어서 구체적으로 작성해드릴게요 — 쓰레드 관리(QThreadPool + QRunnable) 방식으로.
다음 단계로 진행해줘.
좋아, 바로 “자동완성 + 스레드” 단계까지 붙여서 동작하는 전체 예제를 만들어줄게.
아래 코드는 PySide6 기준이며, QThreadPool + QRunnable로 완성 후보 10개를 비동기로 찾고, 입력이 바뀔 때마다 최신 검색만 반영되도록 task_id로 레이스 컨디션을 막았어. 경로 콤보/뒤로·앞으로·상위·새폴더, 파일/폴더 모드, Open/Save, Type 필터, 히스토리, 디바운스까지 들어있어.
import os
import sys
import fnmatch
import traceback
from pathlib import Path
from collections import deque
try:
from PySide6.QtCore import (Qt, QObject, Signal, Slot, QModelIndex,
QTimer, QRunnable, QThreadPool, QMetaObject, Q_ARG)
from PySide6.QtGui import QIcon
from PySide6.QtWidgets import (
QApplication, QDialog, QVBoxLayout, QHBoxLayout, QToolButton, QComboBox,
QListView, QTreeView, QFileSystemModel, QLineEdit, QCompleter,
QDialogButtonBox, QInputDialog, QMessageBox, QLabel, QWidget, QSizePolicy
)
except ImportError:
print("PySide6가 필요합니다. pip install PySide6")
sys.exit(1)
# --------------------------
# 비동기 컴플리터 작업자
# --------------------------
class _CompleterResultEmitter(QObject):
resultsReady = Signal(int, list) # task_id, results 리스트(str)
class PathSearchTask(QRunnable):
"""
입력 문자열을 기반으로 base_dir 아래에서 이름이 prefix로 시작하는 경로(파일/폴더)를
최대 top_k개까지 찾아서 내보내는 작업. (가벼운 BFS, 얕은 탐색)
"""
def __init__(self, task_id: int, base_dir: str, prefix: str,
want_files: bool, want_dirs: bool, top_k: int,
depth_limit: int, emitter: _CompleterResultEmitter):
super().__init__()
self.task_id = task_id
self.base_dir = base_dir
self.prefix = prefix
self.want_files = want_files
self.want_dirs = want_dirs
self.top_k = top_k
self.depth_limit = depth_limit
self.emitter = emitter
def run(self):
results = []
try:
base = Path(self.base_dir).resolve()
if not base.exists():
# 베이스 디렉터리가 없으면 상위들을 타고 올라가며 존재하는 곳부터
for parent in Path(self.base_dir).resolve().parents:
if parent.exists():
base = parent
break
else:
self._emit(results)
return
# 우선 1순위: base_dir의 직접 하위
try:
with os.scandir(base) as it:
for entry in it:
name = entry.name
# 시작 접두사 매칭 (대소문자 구분 없음)
if not name.lower().startswith(self.prefix.lower()):
continue
if entry.is_dir(follow_symlinks=False):
if self.want_dirs:
results.append(str(Path(base, name)))
else:
if self.want_files:
results.append(str(Path(base, name)))
if len(results) >= self.top_k:
self._emit(results)
return
except PermissionError:
pass
# 2순위: 얕은 깊이 탐색 (depth_limit)
# BFS 큐: (Path, depth)
from collections import deque as cdeque
q = cdeque()
q.append((base, 0))
seen = set([str(base)])
while q and len(results) < self.top_k:
cur, depth = q.popleft()
if depth >= self.depth_limit:
continue
try:
with os.scandir(cur) as it:
for entry in it:
try:
full = Path(cur, entry.name)
if entry.is_dir(follow_symlinks=False):
# 디렉터리 자체가 매칭되면 결과에
if self.want_dirs and entry.name.lower().startswith(self.prefix.lower()):
results.append(str(full))
if len(results) >= self.top_k:
self._emit(results)
return
# 하위로 진행
s = str(full)
if s not in seen:
seen.add(s)
q.append((full, depth + 1))
else:
if self.want_files and entry.name.lower().startswith(self.prefix.lower()):
results.append(str(full))
if len(results) >= self.top_k:
self._emit(results)
return
except PermissionError:
continue
except PermissionError:
continue
except Exception:
# 에러는 무시하고 빈 결과
traceback.print_exc()
self._emit(results)
def _emit(self, results):
# UI 스레드로 안전하게 신호
try:
QMetaObject.invokeMethod(
self.emitter,
"resultsReady",
Qt.QueuedConnection,
Q_ARG(int, self.task_id),
Q_ARG(list, results),
)
except Exception:
pass
# --------------------------
# CFileDialog 본체
# --------------------------
class CFileDialog(QDialog):
"""
모드:
mode: "open" | "save"
select_mode: "file" | "folder"
QFileDialog와 유사한 사용법:
dlg = CFileDialog(mode='open', select_mode='file', filters=["All Files (*)", "Images (*.png *.jpg)"])
if dlg.exec():
print(dlg.selected_path())
"""
def __init__(self, start_path=None, mode='open', select_mode='file',
filters=None, parent=None):
super().__init__(parent)
self.setWindowTitle("CFileDialog")
self.resize(900, 600)
self.mode = mode # open/save
self.select_mode = select_mode # file/folder
self.filters = filters or ["All Files (*)"]
self.history_back = deque()
self.history_fwd = deque()
self.current_path = str(Path(start_path).resolve()) if start_path else os.getcwd()
# completer state
self._completer_task_id = 0
self._latest_task_id = 0
self._completer_emitter = _CompleterResultEmitter()
self._completer_emitter.resultsReady.connect(self._on_completer_results)
self._thread_pool = QThreadPool.globalInstance()
self._debounce_timer = QTimer(self)
self._debounce_timer.setInterval(200)
self._debounce_timer.setSingleShot(True)
self._debounce_timer.timeout.connect(self._spawn_completer_task)
self._build_ui()
self._connect_signals()
self._configure_modes()
self._load_path(self.current_path, push_history=False)
# ---------- UI 구성 ----------
def _build_ui(self):
# 상단 툴바
self.back_btn = QToolButton()
self.back_btn.setText("Back")
self.forward_btn = QToolButton()
self.forward_btn.setText("Forward")
self.up_btn = QToolButton()
self.up_btn.setText("Parent")
self.new_folder_btn = QToolButton()
self.new_folder_btn.setText("New Folder")
self.path_combo = QComboBox()
self.path_combo.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
top = QHBoxLayout()
top.addWidget(self.back_btn)
top.addWidget(self.forward_btn)
top.addWidget(self.up_btn)
top.addWidget(self.new_folder_btn)
top.addWidget(self.path_combo)
# 중앙 파일 뷰
self.model = QFileSystemModel(self)
self.model.setRootPath("") # 전체 파일 시스템
# 필터는 모드에 따라 아래에서 설정
self.view = QTreeView() # 트리로 쓰되, 루트 인덱스를 현재 경로로 심어 1계층처럼 사용
self.view.setModel(self.model)
self.view.setRootIsDecorated(False)
self.view.setSortingEnabled(True)
self.view.sortByColumn(0, Qt.AscendingOrder)
self.view.setSelectionBehavior(QTreeView.SelectRows)
self.view.setUniformRowHeights(True)
# 보조 컬럼 정리
for col in range(1, 4):
self.view.setColumnWidth(col, 120)
# 하단: 경로 입력 + Type + 버튼
self.path_edit = QLineEdit()
self.path_edit.setPlaceholderText("경로를 입력하세요…")
self.completer_model = [] # 유지용 (디버그)
self.completer = QCompleter([], self)
self.completer.setCaseSensitivity(Qt.CaseInsensitive)
self.completer.setFilterMode(Qt.MatchContains) # prefix 기반이지만 substring도 보여주면 편함
self.completer.setCompletionMode(QCompleter.PopupCompletion)
self.path_edit.setCompleter(self.completer)
self.type_combo = QComboBox()
self.type_combo.addItems(self.filters)
self.button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel, parent=self)
bottom_row1 = QHBoxLayout()
bottom_row1.addWidget(QLabel("경로:"))
bottom_row1.addWidget(self.path_edit)
bottom_row1.addWidget(QLabel("Type:"))
bottom_row1.addWidget(self.type_combo)
layout = QVBoxLayout(self)
layout.addLayout(top)
layout.addWidget(self.view)
layout.addLayout(bottom_row1)
layout.addWidget(self.button_box)
# ---------- 시그널 연결 ----------
def _connect_signals(self):
self.back_btn.clicked.connect(self._go_back)
self.forward_btn.clicked.connect(self._go_forward)
self.up_btn.clicked.connect(self._go_up)
self.new_folder_btn.clicked.connect(self._create_folder)
self.path_combo.activated.connect(self._combo_jump) # 인덱스 선택 시 이동
self.view.doubleClicked.connect(self._item_double_clicked)
self.view.clicked.connect(self._item_clicked)
self.path_edit.textEdited.connect(self._on_path_edited)
self.type_combo.currentTextChanged.connect(self._on_filter_changed)
self.button_box.accepted.connect(self._accept_if_valid)
self.button_box.rejected.connect(self.reject)
# ---------- 모드/필터 ----------
def _configure_modes(self):
if self.select_mode == 'folder':
# 폴더만
from PySide6.QtCore import QDir
self.model.setFilter(QDir.AllDirs | QDir.NoDotAndDotDot | QDir.Drives)
else:
from PySide6.QtCore import QDir
self.model.setFilter(QDir.AllEntries | QDir.NoDotAndDotDot | QDir.Drives)
# 이름 필터(패턴) 적용
self.model.setNameFilterDisables(False)
self._apply_name_filters_from_combo()
def _apply_name_filters_from_combo(self):
# "Images (*.png *.jpg)" -> ["*.png", "*.jpg"]
text = self.type_combo.currentText()
patterns = []
if "(" in text and ")" in text:
inside = text[text.find("(") + 1:text.find(")")]
patterns = [p.strip() for p in inside.split() if "*" in p or "?" in p]
else:
patterns = ["*"]
if not patterns:
patterns = ["*"]
self.model.setNameFilters(patterns)
def _on_filter_changed(self, _):
self._apply_name_filters_from_combo()
# ---------- 경로 로딩 ----------
def _load_path(self, path: str, push_history=True):
path = str(Path(path).resolve())
if not os.path.exists(path):
return
if push_history:
# 히스토리 푸시
if self.current_path and self.current_path != path:
self.history_back.append(self.current_path)
self.history_fwd.clear()
self.current_path = path
self._update_path_combo(path)
self.path_edit.setText(path)
idx = self.model.index(path)
self.view.setRootIndex(idx)
# OK 버튼 활성화 상태 갱신
self._update_ok_enabled()
def _combo_jump(self, index: int):
target = self.path_combo.itemText(index)
if target:
self._load_path(target)
def _update_path_combo(self, path: str):
self.path_combo.blockSignals(True)
self.path_combo.clear()
p = Path(path).resolve()
parts = p.parts
# 예: /, /a, /a/b, /a/b/c ... 혹은 Windows: C:\, C:\Users, ...
acc = []
for i in range(len(parts)):
acc.append(parts[i])
joined = Path(*acc)
self.path_combo.addItem(str(joined))
# 현재 경로가 가장 아래쪽에 오도록 선택
self.path_combo.setCurrentIndex(self.path_combo.count() - 1)
self.path_combo.blockSignals(False)
# ---------- 탐색 버튼 ----------
def _go_back(self):
if not self.history_back:
return
target = self.history_back.pop()
self.history_fwd.append(self.current_path)
self._load_path(target, push_history=False)
def _go_forward(self):
if not self.history_fwd:
return
target = self.history_fwd.pop()
self.history_back.append(self.current_path)
self._load_path(target, push_history=False)
def _go_up(self):
parent = str(Path(self.current_path).parent)
if parent and parent != self.current_path:
self._load_path(parent)
def _create_folder(self):
name, ok = QInputDialog.getText(self, "새 폴더", "폴더명:")
if not ok or not name.strip():
return
target = Path(self.current_path, name.strip())
try:
target.mkdir(parents=False, exist_ok=False)
# 새 폴더로 포커스 이동
self._load_path(str(target))
except FileExistsError:
QMessageBox.warning(self, "경고", "동일한 이름의 폴더가 이미 존재합니다.")
except PermissionError:
QMessageBox.critical(self, "오류", "폴더를 만들 권한이 없습니다.")
except Exception as e:
QMessageBox.critical(self, "오류", f"폴더 생성 실패: {e}")
# ---------- 뷰 클릭/더블클릭 ----------
def _item_clicked(self, index: QModelIndex):
path = self.model.filePath(index)
# 미리보듯 경로창에 반영
self.path_edit.setText(path)
self._update_ok_enabled()
def _item_double_clicked(self, index: QModelIndex):
path = self.model.filePath(index)
if os.path.isdir(path):
self._load_path(path)
else:
# 파일이면 경로창에 넣고 OK
self.path_edit.setText(path)
self._update_ok_enabled()
if self.mode == 'open' and self.select_mode == 'file':
self.accept()
# ---------- OK 유효성 ----------
def _is_selection_valid(self):
target = self.path_edit.text().strip()
if not target:
return False
if self.select_mode == 'folder':
# 폴더 선택 모드
if os.path.isdir(target):
return True
# Save 모드에서는 새 폴더명도 허용할지? 여기선 폴더 존재만 허용
return False
# 파일 선택 모드
if self.mode == 'open':
# 존재하고, 파일이어야 함
if os.path.isfile(target):
# 필터도 통과해야 함
return self._match_type_filters(Path(target).name)
return False
else:
# save 모드: 존재 여부 무관, 상위 폴더가 있어야 함
parent = Path(target).parent
if parent.exists() and parent.is_dir():
# 파일명 필터도 통과하는지(선택한 필터와 맞지 않아도 저장 허용할지 정책 결정)
# 여기서는 허용(사용자가 필터 바꾸지 않고 확장자 직접 입력 가능)
return True
return False
def _match_type_filters(self, filename: str) -> bool:
text = self.type_combo.currentText()
# 괄호 안 패턴들 추출
patterns = []
if "(" in text and ")" in text:
inside = text[text.find("(") + 1:text.find(")")]
patterns = [p.strip() for p in inside.split() if p.strip()]
if not patterns:
return True
for pat in patterns:
if fnmatch.fnmatch(filename, pat):
return True
return False
def _update_ok_enabled(self):
ok_btn = self.button_box.button(QDialogButtonBox.Ok)
ok_btn.setEnabled(self._is_selection_valid())
def _accept_if_valid(self):
if self._is_selection_valid():
self.accept()
else:
QMessageBox.warning(self, "선택 확인", "선택이 유효하지 않습니다.")
# ---------- 경로 입력 + 비동기 자동완성 ----------
def _on_path_edited(self, _):
# 디바운스 후 검색 태스크 실행
self._debounce_timer.start()
# 입력 중에도 OK 버튼 상태 적절히 조정
self._update_ok_enabled()
def _spawn_completer_task(self):
text = self.path_edit.text().strip()
if not text:
self.completer.model().setStringList([]) if self.completer.model() else self.completer.setModel(QCompleter([], self).model())
return
# base_dir와 prefix 분리
# 예: "/a/b/pa" -> base="/a/b", prefix="pa"
p = Path(text)
if text.endswith(os.sep):
base_dir = text
prefix = ""
else:
base_dir = str(p.parent) if str(p.parent) != "" else os.sep
prefix = p.name
want_files = (self.select_mode == 'file')
want_dirs = True # 항상 디렉터리도 제안: 경로 완성에 중요
top_k = 10
depth_limit = 2 # 너무 깊이 들어가면 느려짐
self._completer_task_id += 1
self._latest_task_id = self._completer_task_id
task = PathSearchTask(self._completer_task_id, base_dir, prefix,
want_files, want_dirs, top_k, depth_limit,
self._completer_emitter)
self._thread_pool.start(task)
@Slot(int, list)
def _on_completer_results(self, task_id: int, results: list):
# 최신 요청만 반영
if task_id != self._latest_task_id:
return
# 결과를 가까운 경로부터 정렬(문자열 길이/사전식)
results_sorted = sorted(results, key=lambda s: (len(s), s.lower()))
# QCompleter 갱신
model = self.completer.model()
if model and hasattr(model, 'setStringList'):
model.setStringList(results_sorted)
else:
string_list_model = QCompleter(results_sorted, self).model()
self.completer.setModel(string_list_model)
self.completer.complete() # 팝업 열기 시도
# ---------- 외부 API ----------
def selected_path(self) -> str:
return self.path_edit.text().strip()
# --------------------------
# 실행 예시
# --------------------------
def main():
app = QApplication(sys.argv)
# 사용 예시 1: 파일 Open
dlg1 = CFileDialog(
start_path=os.getcwd(),
mode='open',
select_mode='file',
filters=[
"All Files (*)",
"Text Files (*.txt *.md)",
"Images (*.png *.jpg *.jpeg *.bmp *.gif)"
]
)
if dlg1.exec():
print("OPEN FILE:", dlg1.selected_path())
# 사용 예시 2: 폴더 선택
dlg2 = CFileDialog(
start_path=os.getcwd(),
mode='open',
select_mode='folder',
filters=["All Files (*)"]
)
if dlg2.exec():
print("SELECT FOLDER:", dlg2.selected_path())
# 사용 예시 3: 파일 Save
dlg3 = CFileDialog(
start_path=os.getcwd(),
mode='save',
select_mode='file',
filters=["All Files (*)", "Python (*.py)"]
)
if dlg3.exec():
print("SAVE FILE:", dlg3.selected_path())
sys.exit(0)
if __name__ == "__main__":
main()
설명 포인트 정리
-
자동완성 비동기화
입력이 바뀌면 200ms 디바운스 후QThreadPool에PathSearchTask를 던져서, 현재 입력 기준base_dir아래에서 이름이 접두사로 맞는 경로들을 최대 10개 찾는다. 결과는task_id로 최신성 확인 후에만QCompleter모델을 갱신한다. 네트워크 드라이브처럼 느린 환경에서도 메인 스레드를 막지 않으면서 점진적으로 후보를 보여준다. -
후보 품질
1순위로 현재 폴더의 직속 항목을 살피고, 부족하면 얕은 깊이(기본 2)만 BFS로 확장해 근처 경로를 제안한다. 정렬은 길이→사전식으로 해서 더 “가까운” 경로가 먼저 온다. -
필터
Type콤보박스의 패턴을QFileSystemModel.setNameFilters로 적용했다. Open 파일 모드에서는 OK 직전에도 파일명이 패턴과 맞는지 빠르게 확인한다. -
동작 정책
Open+File: 존재하는 파일만 OK.
Open+Folder: 존재하는 폴더만 OK.
Save+File: 상위 폴더만 존재하면 OK(파일은 없어도 됨).
Save+Folder 정책은 예제에선 막아두었는데, 원하면 옵션으로 허용 로직을 추가할 수 있다. -
경로 콤보
현재 경로의 루트부터 단계적으로 쌓아서 표시한다. 항목 선택 시 해당 경로로 점프한다. 히스토리 Back/Forward, 상위 이동, 새 폴더도 지원한다.
다음 확장 아이디어
-
네트워크 드라이브에서
depth_limit를 동적으로 조절(지연 길이에 따라 1↔3). -
숨김 파일 토글, 최근 위치/즐겨찾기 사이드바 추가.
-
멀티 선택 지원(파일 다중 선택 모드).
-
Windows 드라이브 루트(“C:\”, “D:\”) 목록을 콤보에 자동 삽입.
추가로 원하는 동작 정책이나 세부 UX가 있으면 말해줘. 바로 코드에 녹여서 완성도를 더 끌어올려볼게.
혹시 더 좋은 개발 방향 있어?
좋지! 지금도 꽤 쓸만하지만, “더 견고하고 빠르고 확장 가능한” 방향으로 다듬을 수 있는 포인트들을 제안할게. 당장 적용할 수 있는 것부터 중장기적인 리팩터링 아이디어까지, 실전적인 순서로 정리했어.
성능·안정성
- 캐싱과 취소 토큰
-
자동완성 후보 탐색 결과를 디렉터리 단위로 캐시하고, mtime(디렉터리 수정 시간) + 사이즈로 신선도 체크.
-
현재 task_id 방식은 최신성 보장은 되지만, 쓰레드가 불필요하게 계속 돈다는 단점이 있어.
PathSearchTask에cancelled플래그(쓰레드 안전) 넣어서 오래 걸리는 네트워크 디렉터리 탐색을 중간 취소 가능하게 만들면 대기시간을 크게 줄일 수 있어.
- 점진적 자동완성(Incremental feed)
- 한 번에 10개를 찾을 때까지 도는 대신, 3개 단위로 먼저 emit → UI 갱신 → 남은 후보 계속 탐색. 사용자는 즉시 반응을 보게 되고, 나머지는 뒤늦게 채워진다.
- QFileSystemWatcher 연동
- 루트 인덱스 기준으로 하위 디렉터리에 변화가 생기면 캐시를 무효화하고 리스트뷰를 부드럽게 갱신. 네트워크 환경에서도 변화를 즉시 반영할 수 있다.
- 아이콘·메타데이터 지연 로딩
QFileSystemModel의 아이콘 로딩은 느릴 수 있다.QFileIconProvider를 상속해 썸네일·아이콘을 별도 스레드에서 지연 로딩하고, 임시 기본 아이콘을 먼저 보여주면 스크롤 지연이 줄어든다.
- 대용량 폴더 가상화
- 수만 개 항목이 있는 디렉터리에서는 모든 항목을 모델에 올리는 것만으로도 느려진다.
QSortFilterProxyModel로 가볍게 필터링하고, 스크롤 위치에 따라 보이는 구간만 프리페치하는 “페이징/윈도우링” 기법을 적용하면 체감 성능이 크게 좋아진다.
아키텍처·유지보수
- 모델/뷰 분리 강화
- 현재는
QFileSystemModel하나로 대부분 해결 중. 여기에 “선택 정책”, “필터 정책”, “히스토리/네비게이션 정책”을 얹은 서비스 계층(작은 클래스로) 분리하면 테스트 용이성↑.
예)NavigationHistory,FilterPolicy,SelectionValidator,CompleterProvider.
- 인터페이스 호환성 레이어
QFileDialog.getOpenFileName()같은 정적 API와 호환되는 래퍼 제공.
예)CFileDialog.getOpenFileName(parent, caption, dir, filter)형태로 반환 타입도 동일하게 맞춰주면 기존 코드 치환이 쉽다.
- 유연한 플러그인 필터
- 단순 패턴 필터(“*.png”) 외에 사용자 정의 필터 콜백을 지원.
예) “최근 24시간 수정” “크기 100MB 이상” 같은 조건을 추가 필터로 끼워 넣기.
- 테스트 전략
- 로컬/네트워크 파일시스템을 흉내내는 가짜 파일시스템(fsys) 어댑터 도입. 비즈니스 로직을 fs 어댑터 인터페이스에 의존하게 하고, 테스트에서는 메모리상 가짜 트리로 검증.
UX·탐색성
- 브레드크럼(breadcrumbs) 위젯
- 현재 경로 콤보는 좋지만, 클릭 가능한 조각(폴더 단추들)로 표시하면 상위로 점프가 더 직관적. 드래그로 하위 펼치기까지 지원하면 금상첨화.
- Places/사이드바
- 홈, 데스크톱, 다운로드, 드라이브 루트, 즐겨찾기(핀), 최근 위치 섹션 추가. Qt의
QFileDialog가 주는 경험을 재현하면서도 커스텀화 가능.
- 키보드 내비게이션
- F2: 이름 바꾸기, Del: 삭제(옵션), Alt+↑: 상위, Alt+←/→: 뒤/앞, Ctrl+L: 경로 포커스 같은 단축키. 텍스트 사용자가 특히 빨라진다.
- 멀티선택
- 파일 모드에서 다중 선택을 옵션으로 지원하고, 하단에 선택 개수/합계 크기 표시. 저장 모드에서도 “여러 개를 한 번에”는 보통 허용하지 않지만 옵션으로 노출 가능.
- 자동완성 UX 디테일
- 탭(Tab)으로 첫 후보 자동완성, 오른쪽 화살표로 경로 구분자 자동 삽입, 끝에 구분자 입력 시 즉시 상위 10개 하위 디렉터리 재검색 등 타자 흐름 최적화.
네트워크·플랫포옴 고려
- UNC/SMB·리눅스 마운트·맥 볼륨
- Windows UNC 경로(\server\share) 파싱과 자동완성, Linux의 /mnt, macOS의 /Volumes 감지. 네트워크 드라이브는 depth_limit을 기본 1로, 지연 시간 측정 후 동적 확장.
- 심볼릭 링크·순환 방지
- BFS에서 symlink는 별도 세트로 추적, 같은 inode/경로 해시 재방문 방지. 네트워크에서 링크 루프가 있을 수 있으니 필수.
- 긴 경로 지원(Windows)
- 260자 제한 우회(\?\ prefix) 처리 옵션.
- 권한·락 대응
- PermissionError는 이미 처리했지만, 저장 모드에서 파일 덮어쓰기 경고, 읽기전용 폴더 알림, 네트워크 타임아웃 후 재시도 버튼 같은 UX를 추가.
접근성·현지화·브랜딩
- 접근성
- 각 버튼에
accessibleName/Description설정, 포커스 인디케이터 가시성 강화, 스크린리더 라벨링.
- 현지화(i18n)
- 문자열 분리와
QTranslator적용 포인트 마련.
- 테마/다크모드
- 아이콘 대비, 선택 하이라이트, 고해상도 스케일링(HiDPI) 보정.
디테일 코드 아이디어
- 취소 가능한 탐색 태스크 스텁
class PathSearchTask(QRunnable):
def __init__(self, ..., cancel_flag):
...
self._cancel_flag = cancel_flag
def run(self):
for entry in slow_scan():
if self._cancel_flag.is_set():
return
...
- 새 요청이 들어오면 이전 태스크의
cancel_flag.set()호출.
- 캐시 키와 무효화
cache_key = (base_dir, want_files, want_dirs)
entries, meta = cache.get(cache_key, (None, None))
if not entries or meta != (mtime(base_dir), dir_size_estimate(base_dir)):
entries = list_dir_fast(base_dir)
cache[cache_key] = (entries, (mtime(...), dir_size_estimate(...)))
- 사용자 정의 필터 콜백
def set_custom_filters(self, predicates: list[Callable[[QFileInfo], bool]]):
self._predicates = predicates
# proxy model의 filterAcceptsRow에서 모든 predicate AND 매칭
- 브레드크럼
QToolButton들을 동적으로 생성하고, 눌렀을 때 그 조각까지의 경로로 이동. 버튼에 메뉴(하위 디렉터리 미리보기) 연결하면 빠른 점프 가능.
단계별 적용 추천
-
1단계: 자동완성 취소·캐시·점진적 갱신
-
2단계: 브레드크럼 + Places 사이드바 + 단축키
-
3단계: 아이콘 지연 로딩 + 대용량 폴더 가상화
-
4단계: 플러그인 필터·테스트 인프라·i18n
원하면 1단계(취소 가능한 자동완성 + 캐시 + 점진 emit) 코드를 네 가지 메소드만 바꿔서 바로 패치 가능한 형태로 제공할게.
Q_ARG에서 에러나
문제 포인트는 여기예요:
QMetaObject.invokeMethod(
self.emitter, "resultsReady", Qt.QueuedConnection,
Q_ARG(int, self.task_id),
Q_ARG(list, results),
)
PySide6에서 Q_ARG(list, …)는 자주 실패합니다. Qt 메타시스템이 Python의 list 타입을 몰라서예요. 또한 이 경우 굳이 QMetaObject.invokeMethod를 쓸 필요가 없습니다. 스레드 간 Signal.emit() 자체가 안전하게 큐드(queued)로 동작합니다.
가장 간단하고 견고한 수정은 다음 두 가지 중 하나입니다.
해결 A: 그냥 시그널 emit으로 바꾸기(권장)
PathSearchTask._emit()을 이렇게 바꾸세요:
def _emit(self, results):
# 워커(쓰레드)에서 emit하면 자동으로 QueuedConnection으로 전달됨
try:
self.emitter.resultsReady.emit(self.task_id, results)
except Exception:
pass
그리고 from PySide6.QtCore import QMetaObject, Q_ARG 임포트는 제거해도 됩니다.
이게 가장 깔끔하고 PySide6에서 권장되는 방식입니다.
해결 B: invokeMethod를 꼭 쓰고 싶다면
list 같은 파이썬 컨테이너는 메타시스템에 등록되어 있지 않아서 타입 자리에 object를 써야 합니다.
from PySide6.QtCore import QMetaObject, Q_ARG
def _emit(self, results):
try:
QMetaObject.invokeMethod(
self.emitter, "resultsReady", Qt.QueuedConnection,
Q_ARG(int, self.task_id),
Q_ARG(object, results), # 핵심: list 대신 object
)
except Exception:
pass
이 방식도 동작하긴 하지만, A안처럼 그냥 emit()을 쓰는 편이 덜 번거롭고 실전에서 오류가 적습니다.
추가로 함께 고치면 좋은 부분
QCompleter에 문자열 목록을 넣을 때는 QStringListModel 하나를 재사용하는 게 좋아요.
from PySide6.QtCore import QStringListModel
# __init__에서
self._slm = QStringListModel([], self)
self.completer = QCompleter(self._slm, self)
self.completer.setCaseSensitivity(Qt.CaseInsensitive)
self.completer.setFilterMode(Qt.MatchContains)
self.completer.setCompletionMode(QCompleter.PopupCompletion)
self.path_edit.setCompleter(self.completer)
# 결과 수신 슬롯에서
@Slot(int, list)
def _on_completer_results(self, task_id: int, results: list):
if task_id != self._latest_task_id:
return
results_sorted = sorted(results, key=lambda s: (len(s), s.lower()))
self._slm.setStringList(results_sorted)
self.completer.complete()
이렇게 하면 모델 재생성에 따른 깜빡임·오버헤드도 줄고, 타입 문제도 사라집니다.
원인 요약: Q_ARG(list, …) 때문에 생긴 타입 미스매치입니다. 스레드-세이프 신호를 직접 emit()하거나, Q_ARG(object, …)로 바꾸면 해결됩니다. 다음에 끊기는 곳이 있으면 에러 메시지 전문을 알려줘요. traceback 기준으로 맞는 패치를 바로 찍어드릴게요.
아니 왜 툴이 멈추는거지 자꾸 /user/ 만 치면 멈춰
멈춤 증상은 보통 세 가지 원인에서 터집니다.
-
비존재/네트워크 경로에서의 과도한 스캔
-
결과 적용 시 메인 스레드 쪽 재진입(Completer 팝업/모델 갱신)
-
이전 태스크가 계속 돌아가는 상태에서 새 태스크가 누적(취소 불가)
아래 패치로 한 번에 잡을 수 있어요. 핵심은 취소 토큰, 안전한 모델 갱신, 루트(/)·빈 접두사에서의 스캔 억제입니다.
1) 취소 가능한 태스크 + QStringListModel 재사용
# __init__ 에 추가
from PySide6.QtCore import QStringListModel
import threading
self._slm = QStringListModel([], self)
self.completer = QCompleter(self._slm, self)
self.completer.setCaseSensitivity(Qt.CaseInsensitive)
self.completer.setFilterMode(Qt.MatchContains)
self.completer.setCompletionMode(QCompleter.PopupCompletion)
self.path_edit.setCompleter(self.completer)
self._cancel_flag = None # 태스크 취소 토큰
2) 빈 접두사/루트 경로에서의 스캔 제한
접두사가 비거나 1글자일 때, 그리고 베이스가 루트일 때는 “직접 하위만” 보거나 아예 스캔을 건너뜁니다. 특히 “/user/” 같이 존재하지 않는 경로를 입력하면 베이스가 “/”로 폴백되는데, 이때 전체 BFS가 급발진할 수 있어서 막아야 해요.
class PathSearchTask(QRunnable):
def __init__(self, task_id, base_dir, prefix, want_files, want_dirs,
top_k, depth_limit, emitter, cancel_flag):
super().__init__()
self.task_id = task_id
self.base_dir = base_dir
self.prefix = prefix
self.want_files = want_files
self.want_dirs = want_dirs
self.top_k = top_k
self.depth_limit = depth_limit
self.emitter = emitter
self.cancel_flag = cancel_flag
def _cancelled(self):
return self.cancel_flag is not None and self.cancel_flag.is_set()
def run(self):
results = []
try:
base = Path(self.base_dir)
# resolve()는 네트워크/비존재 경로에서 멈출 수 있어 생략하고 exists()만 사용
if not base.exists():
# 존재하는 가장 가까운 상위로 폴백
for parent in [*base.parents]:
if parent.exists():
base = parent
break
else:
self._emit(results); return
# 안전장치: 루트에서 접두사 길이 < 2 인 경우는 탐색하지 않음
is_root = str(base) in (os.sep, "C:\\", "D:\\") # 윈도는 필요 시 확장
shallow_only = self.prefix == "" or len(self.prefix) < 2 or is_root
# 1) 직속 항목만 먼저
try:
with os.scandir(base) as it:
for entry in it:
if self._cancelled(): return
name = entry.name
if self.prefix and not name.lower().startswith(self.prefix.lower()):
continue
if entry.is_dir(follow_symlinks=False):
if self.want_dirs:
results.append(str(Path(base, name)))
else:
if self.want_files:
results.append(str(Path(base, name)))
if len(results) >= self.top_k:
self._emit(results); return
except PermissionError:
pass
# 2) 접두사가 충분히 길고 루트가 아니면 얕은 BFS
if not shallow_only:
from collections import deque as cdeque
q = cdeque()
q.append((base, 0))
seen = {str(base)}
while q and len(results) < self.top_k:
if self._cancelled(): return
cur, depth = q.popleft()
if depth >= self.depth_limit:
continue
try:
with os.scandir(cur) as it:
for entry in it:
if self._cancelled(): return
full = Path(cur, entry.name)
if entry.is_dir(follow_symlinks=False):
if self.want_dirs and entry.name.lower().startswith(self.prefix.lower()):
results.append(str(full))
if len(results) >= self.top_k:
self._emit(results); return
s = str(full)
if s not in seen:
seen.add(s); q.append((full, depth + 1))
else:
if self.want_files and entry.name.lower().startswith(self.prefix.lower()):
results.append(str(full))
if len(results) >= self.top_k:
self._emit(results); return
except PermissionError:
continue
finally:
self._emit(results)
def _emit(self, results):
# 스레드에서 emit하면 자동으로 QueuedConnection
try:
self.emitter.resultsReady.emit(self.task_id, results)
except Exception:
pass
3) 새 입력이 오면 이전 태스크 즉시 취소
def _spawn_completer_task(self):
text = self.path_edit.text().strip()
if not text:
self._slm.setStringList([])
return
p = Path(text)
if text.endswith(os.sep):
base_dir = text
prefix = ""
else:
base_dir = str(p.parent) if str(p.parent) else os.sep
prefix = p.name
want_files = (self.select_mode == 'file')
want_dirs = True
top_k = 10
depth_limit = 2
# 이전 태스크 취소
if self._cancel_flag is not None:
self._cancel_flag.set()
self._cancel_flag = threading.Event()
self._completer_task_id += 1
self._latest_task_id = self._completer_task_id
task = PathSearchTask(self._completer_task_id, base_dir, prefix,
want_files, want_dirs, top_k, depth_limit,
self._completer_emitter, self._cancel_flag)
self._thread_pool.start(task)
4) 결과 적용은 가볍게, 팝업 열기는 조건부
팝업을 무조건 열면 입력 중 포커스/선택과 충돌할 수 있어요. 결과가 있을 때, 에디트가 포커스를 가지고 있을 때만 팝업을 열도록 합니다.
@Slot(int, list)
def _on_completer_results(self, task_id: int, results: list):
if task_id != self._latest_task_id:
return
results_sorted = sorted(results, key=lambda s: (len(s), s.lower()))
self._slm.setStringList(results_sorted)
if results_sorted and self.path_edit.hasFocus():
self.completer.complete()
5) 디바운스 상향 + “/user/” 같은 오타 보호
입력이 루트로 폴백될 상황에서 자주 스캔하지 않도록 디바운스를 300~400ms로 올리고, 잘 알려진 루트 후보가 아닌 경우 빈 결과를 줍니다. 예를 들어 macOS는 “/Users”, Linux는 “/usr”, Windows는 “\server\share” 패턴만 허용.
# __init__
self._debounce_timer.setInterval(350)
# _spawn_completer_task 시작부
if text in (os.sep, "/user/", "/usr/", "/Users", "\\\\"): # 필요 시 규칙 확장
# 루트/의심 경로는 빈 후보로
self._slm.setStringList([])
return
이 정도만 반영해도 “/user/” 타이핑 시 전체가 멈추는 현상은 사라질 겁니다. 그래도 멈춘다면 로그를 잠깐 추가해서 어느 단계에서 막히는지 보죠.
import time, logging
logging.basicConfig(level=logging.DEBUG)
# PathSearchTask.run() 주요 단계마다
logging.debug(f"scan base={base} prefix='{self.prefix}' shallow_only={shallow_only}")
# _on_completer_results
logging.debug(f"results[{task_id}]: {len(results)}")
추가 힌트
QFileSystemModel도 대용량 루트에서 초기 스캔이 무겁습니다. 입력창 타이핑만으로 멈춘다면, 뷰의 루트 인덱스를 건드리는 코드가 텍스트 편집 흐름 중에 불려서 모델 프리페치가 겹치는지 확인하세요. 타이핑 중에는 _load_path()가 호출되지 않게(즉, textEdited 경로에서 루트 변경 금지) 한 번 더 점검하면 좋아요.
근데 어디서 freeze가 발생하는건지 모르겠어
가능한 “멈춤 지점”을 빨리 pinpoint하려면, 추측 말고 관측 장비를 꽂아야 해요. 아래 순서대로 넣으면 어디서 멈추는지 10분 안에 알아냅니다.
어디서 얼어붙는지 지도 만들기
관찰 포인트는 보통 이 6곳입니다.
-
textEdited→ 디바운스 타이머 -
_spawn_completer_task진입부 -
PathSearchTask.run()의os.scandir루프 -
워커 → 메인으로 전달(
emit) -
_on_completer_results에서 모델 갱신(QStringListModel.setStringList) -
QCompleter.complete()팝업 열기
여기부터는 바로 붙여 넣을 수 있는 계측 코드예요.
1) 전역 로깅/스톱워치 설정
# 파일 맨 위쪽
import logging, time, threading, signal, faulthandler
logging.basicConfig(
level=logging.DEBUG,
format="%(asctime)s.%(msecs)03d [%(threadName)s] %(levelname)s: %(message)s",
datefmt="%H:%M:%S",
)
# GUI 스레드가 정지했을 때 스택 덤프하기 (리눅스/맥)
try:
faulthandler.register(signal.SIGUSR1)
except Exception:
pass
터미널에서 멈췄다고 느낄 때 리눅스/맥이면 kill -USR1 <pid>를 치면 모든 스레드 스택이 출력돼요. 윈도우라면 아래 단축키로 덤프하게 하면 됩니다.
2) GUI 이벤트 루프가 실제로 멈췄는지 감지(워치독)
# CFileDialog.__init__ 끝에 추가
self._heartbeat = 0
self._watchdog = QTimer(self)
self._watchdog.setInterval(200) # 0.2초마다 heartbeat++
self._watchdog.timeout.connect(lambda: setattr(self, "_heartbeat", self._heartbeat + 1))
self._watchdog.start()
# 1초마다 GUI가 살아있는지 백그라운드 스레드에서 체크
def _watch_gui():
last = -1
while True:
cur = self._heartbeat
time.sleep(1.0)
if cur == self._heartbeat:
logging.error("GUI event loop stalled ≥1s")
last = cur
t = threading.Thread(target=_watch_gui, name="GUI-Watchdog", daemon=True)
t.start()
로그에 “stalled”가 찍히면 진짜 메인스레드가 멈춘 겁니다. 안 찍히면 워커가 오래 걸리거나, complete() 같은 호출이 포커스 충돌로 느려지는 케이스일 가능성이 큽니다.
3) 각 단계 계측 로그
3-1) 입력→태스크 생성
def _on_path_edited(self, _):
logging.debug(f"EDIT text='{self.path_edit.text()}'")
self._debounce_timer.start()
self._update_ok_enabled()
def _spawn_completer_task(self):
text = self.path_edit.text().strip()
logging.debug(f"SPAWN text='{text}'")
start_t = time.perf_counter()
# 기존 취소
if self._cancel_flag is not None:
self._cancel_flag.set()
# base_dir, prefix 계산
p = Path(text)
if text.endswith(os.sep):
base_dir, prefix = text, ""
else:
base_dir = str(p.parent) if str(p.parent) else os.sep
prefix = p.name
logging.debug(f"SPAWN base='{base_dir}' prefix='{prefix}'")
self._cancel_flag = threading.Event()
self._completer_task_id += 1
self._latest_task_id = self._completer_task_id
task = PathSearchTask(self._completer_task_id, base_dir, prefix,
self.select_mode == 'file', True, 10, 2,
self._completer_emitter, self._cancel_flag)
self._thread_pool.start(task)
logging.debug(f"SPAWN dispatched in {(time.perf_counter()-start_t)*1000:.1f} ms")
3-2) 워커 내부
class PathSearchTask(QRunnable):
...
def run(self):
t0 = time.perf_counter()
logging.debug(f"[TASK {self.task_id}] start base='{self.base_dir}' prefix='{self.prefix}'")
results = []
try:
base = Path(self.base_dir) # resolve()는 멈출 수 있어 생략
# 1) 직속
t1 = time.perf_counter()
try:
with os.scandir(base) as it:
for i, entry in enumerate(it):
if self._cancelled():
logging.debug(f"[TASK {self.task_id}] cancelled during scandir")
return
if self.prefix and not entry.name.lower().startswith(self.prefix.lower()):
continue
if entry.is_dir(follow_symlinks=False):
results.append(str(Path(base, entry.name)))
else:
results.append(str(Path(base, entry.name)))
if len(results) >= self.top_k:
break
except PermissionError:
pass
logging.debug(f"[TASK {self.task_id}] direct children took {(time.perf_counter()-t1)*1000:.1f} ms; results={len(results)}")
# 2) BFS는 길면 여기서 오래 걸림
if len(results) < self.top_k and self.prefix and len(self.prefix) >= 2 and str(base) not in (os.sep,):
t2 = time.perf_counter()
from collections import deque as cdeque
q = cdeque([(base,0)])
seen = {str(base)}
while q and len(results) < self.top_k:
if self._cancelled():
logging.debug(f"[TASK {self.task_id}] cancelled during BFS")
return
cur, depth = q.popleft()
if depth >= self.depth_limit:
continue
try:
with os.scandir(cur) as it:
for entry in it:
if self._cancelled(): return
full = Path(cur, entry.name)
if entry.is_dir(follow_symlinks=False):
if entry.name.lower().startswith(self.prefix.lower()):
results.append(str(full))
if len(results) >= self.top_k: break
s = str(full)
if s not in seen:
seen.add(s); q.append((full, depth+1))
else:
if entry.name.lower().startswith(self.prefix.lower()):
results.append(str(full))
if len(results) >= self.top_k: break
except PermissionError:
continue
logging.debug(f"[TASK {self.task_id}] BFS took {(time.perf_counter()-t2)*1000:.1f} ms; results={len(results)}")
finally:
logging.debug(f"[TASK {self.task_id}] total {(time.perf_counter()-t0)*1000:.1f} ms; emit {len(results)}")
self.emitter.resultsReady.emit(self.task_id, results)
3-3) 결과 적용부
@Slot(int, list)
def _on_completer_results(self, task_id: int, results: list):
logging.debug(f"RESULTS task={task_id} latest={self._latest_task_id} count={len(results)} focus={self.path_edit.hasFocus()}")
if task_id != self._latest_task_id:
return
t0 = time.perf_counter()
results_sorted = sorted(results, key=lambda s: (len(s), s.lower()))
self._slm.setStringList(results_sorted)
t1 = time.perf_counter()
if results_sorted and self.path_edit.hasFocus():
self.completer.complete()
logging.debug(f"APPLY setStringList {(t1-t0)*1000:.1f} ms; complete {(time.perf_counter()-t1)*1000:.1f} ms")
4) 루트·의심 경로 차단 규칙 추가
“/user/” 같은 오타가 루트 스캔으로 번지는 걸 방지합니다.
def _spawn_completer_task(self):
text = self.path_edit.text().strip()
if text in (os.sep, "/user/", "/users", "/usr", "\\\\"):
logging.debug("SPAWN short-circuit: suspicious root-ish path")
self._slm.setStringList([])
return
...
5) 수사 가이드라인
로그를 켠 상태에서 직접 타이핑해 보세요.
-
로그가
RESULTS ...까진 잘 나오는데 화면이 버벅이면,setStringList나complete()가 범인입니다.
해결: 후보가 0개일 땐complete()호출 금지, 후보 수를 200개 이상으로 만들지 않기, 팝업이 열려 있을 때만complete()호출. -
TASK ... direct children took ... BFS took ...가 수백~수천 ms로 길면 워커가 네트워크 경로를 훑느라 오래 걸립니다.
해결:depth_limit=1, prefix 길이 2 미만이면 BFS 금지, 혹은 “직속 하위만”으로 제한. -
“GUI event loop stalled ≥1s”가 뜨면 메인 스레드가 막힌 겁니다. 이때 SIGUSR1(리눅스/맥)이나 아래 단축키로 스택을 덤프해 원인을 봅니다.
6) 모든 스레드 스택을 단축키로 덤프(윈도우 포함)
# CFileDialog._build_ui() 마지막쯤
QShortcut = __import__("PySide6.QtGui", fromlist=["QShortcut"]).QShortcut
QKeySequence = __import__("PySide6.QtGui", fromlist=["QKeySequence"]).QKeySequence
dump_shortcut = QShortcut(QKeySequence("F12"), self)
def _dump():
logging.error("=== STACK DUMP REQUESTED ===")
try:
faulthandler.dump_traceback(all_threads=True)
except Exception as e:
logging.error(f"faulthandler dump failed: {e}")
dump_shortcut.activated.connect(_dump)
F12를 누르면 현재 모든 스레드의 파이썬 스택이 출력됩니다. 거기에 QFileSystemModel.index(), QCompleter.complete(), os.scandir() 등이 보이면 범인이 명확해져요.
이 계측을 넣고 타이핑해 보면, 멈춤이 1) 워커 과부하인지, 2) 결과 적용(모델/컴플리터)에서 포커스 재진입 문제인지, 3) 루트 스캔 폭주인지 바로 드러납니다. 로그 몇 줄만 붙여 주면, 그에 맞춰 해당 지점을 수술해서 완전히 고정해 드릴게요.
너무 많이 찾을까봐 10개로 줄였잖아
맞아요, “10개만”으로 제한해도 멈출 수 있어요. 이유는 “개수”가 아니라 “나열(열기) 자체”가 막히기 때문이에요. 네트워크/루트 경로에서 os.scandir()(혹은 존재 확인) 자체가 블로킹되면 결과를 1개만 원해도 그 1개를 얻기 전에 스레드가 붙잡혀요. 심지어 Path(...).resolve()도 원격 경로에서 멈출 수 있습니다.
아래처럼 바꾸면 멈춤 원인을 사실상 제거할 수 있습니다. 변화량은 작고 효과는 큽니다.
멈춤을 유발하는 지점
-
루트(
/,C:\,\\server\) 또는 잘못된 상위에서scandir()시도 -
resolve()로 네트워크/심볼릭 링크 추적 -
짧은 접두사(예: “/us”)일 때 광범위 탐색
-
이전 작업을 취소해도 실제
scandir()가 이미 막혔으면 스레드가 오래 살아있음(새 입력과 경합)
안전장치 6종 세트
- 절대
resolve()금지
자동완성 워커에서는 경로 정규화가 필요 없습니다. 존재 확인만 느슨하게:
# base = Path(self.base_dir).resolve() # 금지
base = Path(self.base_dir) # 그대로 둠
# 존재 확인도 느리면 건너뜀
exists_try = base.exists() if len(str(base)) < 260 else True
- “루트 + 짧은 접두사”는 스캔하지 않기
루트에서 길이 0~1 접두사는 거의 항상 과탐색을 유도합니다.
is_unc = str(base).startswith("\\\\")
is_root = str(base) in (os.sep, "C:\\", "D:\\") or is_unc or str(base).startswith("/Volumes")
short_prefix = (self.prefix == "" or len(self.prefix) < 2)
if is_root and short_prefix:
self._emit([]); return
- 직속 항목만 “가능할 때만” 확인
scandir()호출 자체가 막히면 답이 없으니, 먼저 “짧은 시간 예열” 후 넘깁니다.
import time
t0 = time.perf_counter()
try:
with os.scandir(base) as it:
for entry in it:
if self._cancelled(): return
# … 필터링 …
if len(results) >= self.top_k: break
if (time.perf_counter() - t0) > 0.05: # 50ms 이상 쓰면 중단
break
except PermissionError:
pass
except OSError:
# 원격 타임아웃/연결 문제 등
self._emit(results); return
시간 예산을 50~100ms로 두면 “타자 중” 응답성이 유지됩니다.
- BFS 완전 금지(혹은 매우 엄격히)
자동완성에서 BFS는 체감가치 대비 리스크가 큽니다. 최소 요건을 이렇게:
if not (not is_root and len(self.prefix) >= 3):
# BFS 생략
self._emit(results); return
# 그래도 하려면 depth_limit=1, 그리고 동일한 50ms 타임박스 내로
- 별도 스레드풀 + 단일 스레드
이전 작업이 막히더라도 새 작업이 또 채워지지 않도록 자동완성 전용 풀을 만들어 1개만 돌립니다.
# __init__
self._auto_pool = QThreadPool(self)
self._auto_pool.setMaxThreadCount(1) # 동시에 1개만
# _spawn_completer_task 끝
self._auto_pool.start(task) # 기존 globalInstance() 대신
- 입력 정책: “확실할 때만 자동완성”
네트워크 경로/루트에서 접두사 2글자 미만이면 아예 후보를 내지 않습니다. UX적으로도 납득이 됩니다.
# _spawn_completer_task 초반
text = self.path_edit.text().strip()
if text in (os.sep, "\\\\") or text.endswith(os.sep) and len(Path(text).name) == 0:
self._slm.setStringList([]); return
# UNC나 /Volumes, /mnt, /net 등에서 prefix<2면 스킵
s = text.lower()
if s.startswith("\\\\") or s.startswith("/volumes") or s.startswith("/mnt") or s.startswith("/net"):
tail = Path(text).name
if len(tail) < 2:
self._slm.setStringList([]); return
작게 고쳐도 큰 효과가 나는 포인트
QCompleter갱신은 최대 10개라도, 팝업을 매번 열면 메인 스레드가 레이아웃·측정으로 바빠집니다. 결과가 비었으면complete()호출 금지, 이미 열려 있으면만 갱신:
@Slot(int, list)
def _on_completer_results(self, task_id, results):
if task_id != self._latest_task_id: return
items = sorted(results, key=lambda s:(len(s), s.lower()))[:10]
self._slm.setStringList(items)
if items and self.path_edit.hasFocus() and self.completer.popup().isVisible():
self.completer.complete()
-
textEdited에서 절대_load_path()를 호출하지 말 것. 입력 중 루트 인덱스 변경은QFileSystemModel의 백그라운드 스캔을 유발해 같이 굳습니다. -
exists()도 원격에서 느릴 수 있으니, 자동완성 워커에서는 사용을 최소화하고, 최종 OK 검증 때만 엄격히 확인하세요.
한 번에 적용하는 최소 패치 요약
-
워커에서
resolve()제거 -
루트+짧은 접두사면 바로 빈 결과 반환
-
scandir()에 50~100ms 타임박스 -
BFS 끄기(혹은 prefix≥3 & depth=1 & 타임박스)
-
자동완성 전용
QThreadPool(max=1)사용 -
complete()는 팝업 열려 있을 때만
이렇게만 바꿔도 “/user/만 치면 멈춤” 류가 사라집니다.
개수 제한은 “열거가 끝난 뒤”에나 의미가 있으니, 열거 자체가 막히지 않게 설계를 틀어주는 게 핵심입니다.
근데 쓰레드에서 요청하는건데 왜 gui도 freeze 되는거임
짚고 넘어가자면, “워커 스레드에서 돌리는데도 GUI가 얼어붙는” 건 꽤 흔한 현상이에요. 원인은 보통 아래 네 가지 중 하나(혹은 복합)입니다. 핵심은 GUI가 진짜 멈춘 게 아니라, 메인 스레드가 일을 못 하거나 대기 상태에 빠져 “멈춘 것처럼 보이는” 겁니다.
왜 멈추는가
-
GIL(파이썬 전역 락) 점유로 메인 스레드가 굶주림
워커 스레드가 파일 시스템을 훑으면서 Python 레벨 루프를 오래 돌면 GIL을 오래 쥐고 있게 됩니다. Qt의 이벤트 루프 자체는 C++이라 돌아가지만, 메인 스레드에서 실행되는 Python 슬롯/타이머/paint 이벤트 처리에는 GIL이 필요합니다. 워커가 GIL을 오래 잡고 있으면 메인 쪽 Python 코드가 실행되지 못해 “프리즈처럼” 보입니다.
증상: 타이머 콜백, 슬롯이 늦게 오거나 아예 멈춘 것처럼 보임. -
메인 스레드 쪽에서 무거운 작업이 실행됨
신호는 큐드(queued)로 잘 왔는데, 결과를 적용하는 슬롯에서 무거운 일을 하면(대량 리스트를QStringListModel.setStringList()로 교체,QCompleter.complete()반복 호출,QFileSystemModel루트 변경 등) 그 순간만큼 메인 스레드가 바빠 보입니다. 특히 팝업 열기와 레이아웃 재계산은 생각보다 비쌉니다. -
Qt 객체를 워커 스레드에서 건드림
Qt 위젯/모델은 “그 객체가 속한 스레드”에서만 만져야 합니다. 워커 내부에서 Qt 객체를 직접 건드리면 undefined behavior → 교착/프리즈. 반드시 워커는 파이썬의 순수 데이터(list[str])만 만들고, 메인 슬롯에서만 Qt 객체를 수정하세요. -
I/O가 “스레드가 아니라 OS 레벨”에서 막힘
네트워크 경로, UNC, 루트 등에서Path.resolve()나os.scandir()첫 호출이 OS 쪽에서 오래 블록될 수 있습니다. 스레드에서 막히는 것이 맞지만, 그 사이 결과를 기다리며 메인에서 계속 팝업을 열고 닫고, 모델을 비우고 채우는 작업이 연쇄되면 체감상 전체가 굼벵이처럼 느립니다.
확인 체크리스트
-
워커 코드가 CPU 루프를 오래 잡지 않게 타임박스 적용(예:
scandir50~100ms 한도). -
워커 루프 중 주기적으로
time.sleep(0)또는 아주 짧은time.sleep(0.001)넣어 GIL 양보.sleep()은 GIL을 놓습니다. -
워커에서는 절대 Qt 객체 접근 금지. 신호로 순수 파이썬 자료만 전달.
-
메인 슬롯에서의 작업을 O(候補수)로 유지.
setStringList후 즉시complete()를 매번 호출하지 말고, 팝업이 이미 열려 있을 때만, 결과가 변했을 때만 갱신. -
입력 중에는
QFileSystemModel루트 변경 금지. 루트 변경은 별도 액션(엔터, 버튼, 더블클릭)에서만.
바로 적용 가능한 수습 팁
- 워커 루프 타임박스와 GIL 양보:
import time
t0 = time.perf_counter()
for entry in it: # scandir 루프
# ... 필터링/수집 ...
if (time.perf_counter() - t0) > 0.05: # 50ms 예산
break
if i % 50 == 0:
time.sleep(0) # GIL 양보
-
결과 적용 최소화:
-
QStringListModel재사용(이미 하셨다면 유지). -
이전 결과와 새 결과가 같으면 아무것도 하지 않기.
-
팝업이 열려 있을 때만
complete()호출. -
후보가 0개면
complete()호출 금지.
-
-
루트·짧은 접두사 스킵: 루트(‘/’, ‘\’)에서 prefix 길이 < 2면 아예 탐색하지 않고 빈 후보 반환.
-
자동완성 전용
QThreadPool을 1스레드로 제한해서 태스크 적체 방지:
self._auto_pool = QThreadPool(self)
self._auto_pool.setMaxThreadCount(1)
# self._auto_pool.start(task)
“왜”의 핵심 요약
-
스레드라서 GUI가 절대 안 멈추는 게 아닙니다.
파이썬은 GIL 때문에 워커가 오래 점유하면 메인의 Python 코드가 멈춘 듯 보입니다. -
신호가 메인에 도착해도, 그걸 처리하는 슬롯이 무겁거나 자주 불리면 메인이 바빠져서 역시 “멈춘 느낌”이 납니다.
-
Qt 객체는 메인에서만 만지세요. 워커는 데이터만 만들고, 짧게, 타임박스 안에서, GIL을 주기적으로 놓으면서 돌리면 체감 프리즈가 사라집니다.
원하면 당신 코드의 워커 부분과 결과 적용 슬롯 부분을 딱 10줄 내로 패치한 버전을 만들어서 붙여드릴게요.
그러면 쓰레드 말고 멀티 프로세싱으로 가자
좋아, 자동완성만 “멀티프로세싱”으로 빼서 GIL 영향을 원천 차단하는 쪽으로 정리하자. 핵심은 다음 셋업이야.
-
워커는 프로세스. 입력 문자열을 받아서 경로 후보 10개만 계산해 순수 파이썬 리스트로 반환.
-
GUI는 여전히 메인 스레드에서만 Qt 객체를 만짐. 결과는 시그널로만 반영.
-
위험 구간(루트·짧은 접두사·네트워크 루트)의 탐색을 타임박스(예: 50~100ms)로 제한.
아래는 너의 기존 코드에서 “컴플리터 스레드” 부분만 교체하면 되는 드롭인 패치야. PySide6 그대로, ProcessPoolExecutor(max_workers=1)를 사용해 요청을 순서대로 처리하고, 이전 요청은 task_id로 무시한다.
0) 주의사항
-
Windows/macOS에서 프로세스는 spawn이 기본이라, 워커 함수는 반드시 모듈 최상위에 있어야 하고(피클링),
if __name__ == "__main__":가드 안에서 앱 실행해야 해. 아래 예제는 그 규칙을 지킨 형태야. -
Qt 객체는 절대 워커에서 접근하지 말 것. 경로 문자열만 주고 문자열 리스트만 받는다.
-
프로세스 쪽에서
resolve()는 금지. 네트워크에서 막힐 수 있다.
1) 모듈 최상위: 워커 함수 추가
# --- 모듈 최상위에 추가 ---
import os, time
from pathlib import Path
def _collect_completions(base_dir: str, prefix: str, want_files: bool, want_dirs: bool,
top_k: int = 10, depth_limit: int = 1,
time_budget_ms: int = 80) -> list[str]:
"""
멀티프로세스 워커. 순수 함수.
- 루트 + 짧은 접두사면 바로 빈 리스트.
- 직속 항목만 우선, 필요시 얕은 BFS(depth<=1)로 보강.
- scandir 루프는 타임박스(기본 80ms) 내에서만 실행.
"""
try:
base = Path(base_dir)
sbase = str(base)
is_unc = sbase.startswith("\\\\")
is_root = sbase in (os.sep, "C:\\", "D:\\", "E:\\") or is_unc or sbase.startswith("/Volumes")
short_prefix = (prefix == "" or len(prefix) < 2)
if is_root and short_prefix:
return []
# 존재 확인은 느릴 수 있으므로 무조건 강행하지 말고 가볍게 시도
try:
exists = base.exists()
except Exception:
exists = False
if not exists:
# 가까운 상위로 폴백
for parent in base.parents:
try:
if parent.exists():
base = parent
break
except Exception:
continue
results = []
budget = time_budget_ms / 1000.0
t0 = time.perf_counter()
# 1) 직속
try:
with os.scandir(base) as it:
for entry in it:
name = entry.name
if prefix and not name.lower().startswith(prefix.lower()):
if (time.perf_counter() - t0) > budget:
break
continue
if entry.is_dir(follow_symlinks=False):
if want_dirs:
results.append(str(Path(base, name)))
else:
if want_files:
results.append(str(Path(base, name)))
if len(results) >= top_k or (time.perf_counter() - t0) > budget:
break
except Exception:
pass
# 2) 보강 BFS(아주 얕게, 조건부)
if len(results) < top_k and not is_root and len(prefix) >= 3 and (time.perf_counter() - t0) <= budget:
from collections import deque as cdeque
q = cdeque([(base, 0)])
seen = {str(base)}
while q and len(results) < top_k and (time.perf_counter() - t0) <= budget:
cur, depth = q.popleft()
if depth >= depth_limit:
continue
try:
with os.scandir(cur) as it:
for entry in it:
full = Path(cur, entry.name)
if entry.is_dir(follow_symlinks=False):
if entry.name.lower().startswith(prefix.lower()) and want_dirs:
results.append(str(full))
if len(results) >= top_k or (time.perf_counter() - t0) > budget:
break
s = str(full)
if s not in seen:
seen.add(s); q.append((full, depth + 1))
else:
if entry.name.lower().startswith(prefix.lower()) and want_files:
results.append(str(full))
if len(results) >= top_k or (time.perf_counter() - t0) > budget:
break
except Exception:
continue
# 정렬은 짧은 경로 우선
results = sorted(results, key=lambda s: (len(s), s.lower()))[:top_k]
return results
except Exception:
return []
2) 다이얼로그: 프로세스 풀과 시그널 준비
# __init__ 안에 (기존 스레드풀·워커 관련 코드 제거하고) 추가
from concurrent.futures import ProcessPoolExecutor
from PySide6.QtCore import QObject, Signal, Slot, QTimer, QStringListModel
self._slm = QStringListModel([], self)
self.completer.setModel(self._slm)
self._proc_pool = ProcessPoolExecutor(max_workers=1) # 자동완성 전용 1프로세스
self._pending_future = None
self._task_seq = 0
# 결과를 Qt로 넘길 emitter
class _ProcEmitter(QObject):
resultsReady = Signal(int, list)
self._proc_emitter = _ProcEmitter()
self._proc_emitter.resultsReady.connect(self._on_proc_results)
다이얼로그가 닫힐 때 풀을 정리해 줘야 해.
def closeEvent(self, e):
try:
if self._pending_future:
self._pending_future.cancel()
self._proc_pool.shutdown(wait=False, cancel_futures=True)
except Exception:
pass
super().closeEvent(e)
3) 작업 투입 로직 교체
def _spawn_completer_task(self):
text = self.path_edit.text().strip()
if not text:
self._slm.setStringList([])
return
p = Path(text)
if text.endswith(os.sep):
base_dir = text
prefix = ""
else:
base_dir = str(p.parent) if str(p.parent) else os.sep
prefix = p.name
# 루트·의심 경로 단축 리턴
sbase = base_dir.lower()
if base_dir in (os.sep, "\\\\") or sbase.startswith("/volumes") or sbase.startswith("/mnt") or sbase.startswith("/net"):
if len(prefix) < 2:
self._slm.setStringList([])
return
want_files = (self.select_mode == 'file')
want_dirs = True
# 이전 요청 무시(취소 시도)
if self._pending_future and not self._pending_future.done():
self._pending_future.cancel() # 이미 실행 중이면 취소가 보장되진 않지만 결과는 무시할 거라 OK
self._task_seq += 1
seq = self._task_seq
# 제출
fut = self._proc_pool.submit(_collect_completions, base_dir, prefix, want_files, want_dirs, 10, 1, 80)
self._pending_future = fut
# 완료 콜백: 워커 스레드에서 실행→Qt 시그널로 메인에 전달
def _on_done(f):
try:
results = f.result(timeout=0) if f.done() else []
except Exception:
results = []
self._proc_emitter.resultsReady.emit(seq, results)
fut.add_done_callback(_on_done)
4) 결과 적용 슬롯
@Slot(int, list)
def _on_proc_results(self, seq: int, results: list):
if seq != self._task_seq:
return # 최신 요청만 반영
# 경량 갱신
items = results[:10]
# 이전과 동일하면 갱신 생략
if self._slm.stringList() == items:
return
self._slm.setStringList(items)
# 팝업은 필요할 때만
if items and self.path_edit.hasFocus() and self.completer.popup().isVisible():
self.completer.complete()
5) 엔트리 포인트 가드
멀티프로세싱을 쓰면 메인 진입을 가드해야 해.
def main():
app = QApplication(sys.argv)
# ... 다이얼로그 실행 ...
if __name__ == "__main__":
# 필요시: from multiprocessing import set_start_method; set_start_method("spawn", force=True)
main()
6) 왜 이 구성이 안전한가
-
GIL: 후보 수집은 별도 프로세스에서 돌아서 메인의 파이썬 슬롯·타이머가 굶지 않는다.
-
Qt 스레드 규칙: 워커는 순수 데이터만 반환하고, Qt 객체 갱신은 시그널 슬롯에서만 처리.
-
블로킹 I/O: 네트워크에서
scandir()가 잠시 멈춰도 GUI는 독립적. 타임박스로 장기 블로킹을 더 줄였다. -
폭주 방지:
max_workers=1로 큐 누적을 제한하고, task_seq로 오래된 결과를 무시한다.
확장 팁
-
네트워크 드라이브에서 더 여유가 필요하면
time_budget_ms를 120~150으로, 대신depth_limit=0으로 유지(직속만). -
macOS 앱 번들로 배포할 때는 워커 함수를 별도
.py모듈로 분리하면 스폰 안정성이 올라간다. -
후보 캐싱을 추가하면 같은
base_dir에서 접두사가 길어질수록 재사용 가능(프로세스 메모리에도 캐시를 둘지, 메인에서 최근 결과를 재활용할지 선택).
이대로 붙여서 돌려보고, 여전히 특정 경로에서 체감 지연이 있으면 로그 한두 줄만 공유해줘. 타임박스·깊이·루트 규칙을 경로별로 미세 조정해 더 민첩하게 만들 수 있어.
전체 코드 좀
좋아. 아래는 멀티프로세싱 기반 자동완성(컴플리터)을 포함한 전체 동작 예제야.
PySide6 기준이고, ProcessPoolExecutor(max_workers=1)로 자동완성만 별도 프로세스에서 처리해서 GIL·블로킹 영향을 차단해. 경로 콤보/Back·Forward·Parent·New Folder, 파일·폴더 모드, Open/Save, Type 필터, 히스토리, 디바운스, 그리고 안전장치(루트+짧은 접두사 차단, scandir() 타임박스)까지 포함했어.
# -*- coding: utf-8 -*-
import os
import sys
import fnmatch
import time
from collections import deque
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor
# ==========================================
# 멀티프로세스 워커 함수 (모듈 최상위에 있어야 함)
# ==========================================
def _collect_completions(base_dir: str, prefix: str,
want_files: bool, want_dirs: bool,
top_k: int = 10, depth_limit: int = 1,
time_budget_ms: int = 80) -> list[str]:
"""
자동완성 후보를 수집하는 순수 함수(별도 프로세스에서 실행).
- 루트 + 짧은 접두사면 빈 리스트 반환
- 직속 항목 우선, 필요 시 매우 얕은 BFS(depth<=1)로 보강
- os.scandir 루프는 타임박스 내에서만 동작
- Qt 객체 접근 절대 금지 (문자열 리스트만 반환)
"""
try:
base = Path(base_dir)
sbase = str(base)
# 플랫폼별 루트 판단
is_unc = sbase.startswith("\\\\")
is_root_like = (
sbase == os.sep or
sbase in ("C:\\", "D:\\", "E:\\", "F:\\") or
is_unc or
sbase.startswith("/Volumes") or
sbase == "/"
)
short_prefix = (prefix == "" or len(prefix) < 2)
if is_root_like and short_prefix:
return []
# 존재 확인은 네트워크에서 느릴 수 있으니 조심스럽게
try:
exists = base.exists()
except Exception:
exists = False
if not exists:
# 가까운 상위로 폴백
for parent in base.parents:
try:
if parent.exists():
base = parent
break
except Exception:
continue
results = []
budget = time_budget_ms / 1000.0
t0 = time.perf_counter()
# 1) 직속 항목 스캔 (타임박스)
try:
with os.scandir(base) as it:
for entry in it:
name = entry.name
if prefix and not name.lower().startswith(prefix.lower()):
if (time.perf_counter() - t0) > budget:
break
continue
if entry.is_dir(follow_symlinks=False):
if want_dirs:
results.append(str(Path(base, name)))
else:
if want_files:
results.append(str(Path(base, name)))
if len(results) >= top_k or (time.perf_counter() - t0) > budget:
break
except Exception:
# PermissionError, OSError 등은 무시
pass
# 2) 보강 BFS (조건부, 매우 얕게, 타임박스)
if (len(results) < top_k
and not is_root_like
and len(prefix) >= 3
and (time.perf_counter() - t0) <= budget
and depth_limit > 0):
from collections import deque as cdeque
q = cdeque([(base, 0)])
seen = {str(base)}
while q and len(results) < top_k and (time.perf_counter() - t0) <= budget:
cur, depth = q.popleft()
if depth >= depth_limit:
continue
try:
with os.scandir(cur) as it:
for entry in it:
full = Path(cur, entry.name)
# 매칭 확인
if entry.is_dir(follow_symlinks=False):
if entry.name.lower().startswith(prefix.lower()) and want_dirs:
results.append(str(full))
if len(results) >= top_k or (time.perf_counter() - t0) > budget:
break
s = str(full)
if s not in seen:
seen.add(s)
q.append((full, depth + 1))
else:
if entry.name.lower().startswith(prefix.lower()) and want_files:
results.append(str(full))
if len(results) >= top_k or (time.perf_counter() - t0) > budget:
break
except Exception:
continue
results = sorted(results, key=lambda s: (len(s), s.lower()))[:top_k]
return results
except Exception:
return []
# ==========================================
# Qt 임포트
# ==========================================
try:
from PySide6.QtCore import Qt, QObject, Signal, Slot, QModelIndex, QTimer, QStringListModel
from PySide6.QtWidgets import (
QApplication, QDialog, QVBoxLayout, QHBoxLayout, QToolButton, QComboBox,
QTreeView, QFileSystemModel, QLineEdit, QCompleter, QDialogButtonBox,
QInputDialog, QMessageBox, QLabel, QSizePolicy
)
except ImportError:
print("PySide6가 필요합니다. pip install PySide6")
sys.exit(1)
# ==========================================
# CFileDialog 본체
# ==========================================
class CFileDialog(QDialog):
"""
mode: 'open' | 'save'
select_mode: 'file' | 'folder'
사용 예:
dlg = CFileDialog(mode='open', select_mode='file',
filters=["All Files (*)", "Images (*.png *.jpg)"])
if dlg.exec():
print(dlg.selected_path())
"""
def __init__(self, start_path=None, mode='open', select_mode='file',
filters=None, parent=None):
super().__init__(parent)
self.setWindowTitle("CFileDialog (Multiprocessing Completer)")
self.resize(960, 640)
self.mode = mode
self.select_mode = select_mode
self.filters = filters or ["All Files (*)"]
self.history_back = deque()
self.history_fwd = deque()
self.current_path = str(Path(start_path).resolve()) if start_path else os.getcwd()
# 자동완성 상태
self._debounce_timer = QTimer(self)
self._debounce_timer.setInterval(300) # 입력 디바운스
self._debounce_timer.setSingleShot(True)
self._debounce_timer.timeout.connect(self._spawn_completer_task)
# 프로세스 풀: 자동완성 전용, 동시 1개
self._proc_pool = ProcessPoolExecutor(max_workers=1)
self._pending_future = None
self._task_seq = 0
# 프로세스 결과를 Qt로 전달하는 emitter
class _ProcEmitter(QObject):
resultsReady = Signal(int, list)
self._proc_emitter = _ProcEmitter()
self._proc_emitter.resultsReady.connect(self._on_proc_results)
self._build_ui()
self._connect_signals()
self._configure_modes()
self._load_path(self.current_path, push_history=False)
# ---------- UI 구성 ----------
def _build_ui(self):
# 상단 툴바
self.back_btn = QToolButton(text="Back")
self.forward_btn = QToolButton(text="Forward")
self.up_btn = QToolButton(text="Parent")
self.new_folder_btn = QToolButton(text="New Folder")
self.path_combo = QComboBox()
self.path_combo.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
top = QHBoxLayout()
top.addWidget(self.back_btn)
top.addWidget(self.forward_btn)
top.addWidget(self.up_btn)
top.addWidget(self.new_folder_btn)
top.addWidget(self.path_combo)
# 중앙 뷰
self.model = QFileSystemModel(self)
self.model.setRootPath("") # 전체 FS
self.view = QTreeView()
self.view.setModel(self.model)
self.view.setRootIsDecorated(False)
self.view.setSortingEnabled(True)
self.view.sortByColumn(0, Qt.AscendingOrder)
self.view.setSelectionBehavior(QTreeView.SelectRows)
self.view.setUniformRowHeights(True)
for col in range(1, 4):
self.view.setColumnWidth(col, 120)
# 하단: 경로 입력 + Type + 버튼들
self.path_edit = QLineEdit()
self.path_edit.setPlaceholderText("경로를 입력하세요…")
# QCompleter + QStringListModel 재사용
self._slm = QStringListModel([], self)
self.completer = QCompleter(self._slm, self)
self.completer.setCaseSensitivity(Qt.CaseInsensitive)
self.completer.setFilterMode(Qt.MatchContains)
self.completer.setCompletionMode(QCompleter.PopupCompletion)
self.path_edit.setCompleter(self.completer)
self.type_combo = QComboBox()
self.type_combo.addItems(self.filters)
self.button_box = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel, parent=self)
bottom = QHBoxLayout()
bottom.addWidget(QLabel("경로:"))
bottom.addWidget(self.path_edit)
bottom.addWidget(QLabel("Type:"))
bottom.addWidget(self.type_combo)
layout = QVBoxLayout(self)
layout.addLayout(top)
layout.addWidget(self.view)
layout.addLayout(bottom)
layout.addWidget(self.button_box)
# ---------- 시그널 ----------
def _connect_signals(self):
self.back_btn.clicked.connect(self._go_back)
self.forward_btn.clicked.connect(self._go_forward)
self.up_btn.clicked.connect(self._go_up)
self.new_folder_btn.clicked.connect(self._create_folder)
self.path_combo.activated.connect(self._combo_jump)
self.view.doubleClicked.connect(self._item_double_clicked)
self.view.clicked.connect(self._item_clicked)
self.path_edit.textEdited.connect(self._on_path_edited)
self.type_combo.currentTextChanged.connect(self._on_filter_changed)
self.button_box.accepted.connect(self._accept_if_valid)
self.button_box.rejected.connect(self.reject)
# ---------- 모드/필터 ----------
def _configure_modes(self):
from PySide6.QtCore import QDir
if self.select_mode == 'folder':
self.model.setFilter(QDir.AllDirs | QDir.NoDotAndDotDot | QDir.Drives)
else:
self.model.setFilter(QDir.AllEntries | QDir.NoDotAndDotDot | QDir.Drives)
self.model.setNameFilterDisables(False)
self._apply_name_filters_from_combo()
def _apply_name_filters_from_combo(self):
text = self.type_combo.currentText()
patterns = []
if "(" in text and ")" in text:
inside = text[text.find("(") + 1:text.find(")")]
patterns = [p.strip() for p in inside.split() if p.strip()]
if not patterns:
patterns = ["*"]
self.model.setNameFilters(patterns)
def _on_filter_changed(self, _):
self._apply_name_filters_from_combo()
# ---------- 경로 로딩 ----------
def _load_path(self, path: str, push_history=True):
# resolve()는 메인에서만 사용 (UI 반응 느려지면 생략해도 됨)
try:
path = str(Path(path).resolve())
except Exception:
path = str(Path(path))
if not os.path.exists(path):
return
if push_history and self.current_path and self.current_path != path:
self.history_back.append(self.current_path)
self.history_fwd.clear()
self.current_path = path
self._update_path_combo(path)
self.path_edit.setText(path)
self.view.setRootIndex(self.model.index(path))
self._update_ok_enabled()
def _combo_jump(self, index: int):
target = self.path_combo.itemText(index)
if target:
self._load_path(target)
def _update_path_combo(self, path: str):
self.path_combo.blockSignals(True)
self.path_combo.clear()
p = Path(path)
parts = p.parts
acc = []
for i in range(len(parts)):
acc.append(parts[i])
self.path_combo.addItem(str(Path(*acc)))
self.path_combo.setCurrentIndex(self.path_combo.count() - 1)
self.path_combo.blockSignals(False)
# ---------- 탐색 ----------
def _go_back(self):
if not self.history_back:
return
target = self.history_back.pop()
self.history_fwd.append(self.current_path)
self._load_path(target, push_history=False)
def _go_forward(self):
if not self.history_fwd:
return
target = self.history_fwd.pop()
self.history_back.append(self.current_path)
self._load_path(target, push_history=False)
def _go_up(self):
parent = str(Path(self.current_path).parent)
if parent and parent != self.current_path:
self._load_path(parent)
def _create_folder(self):
name, ok = QInputDialog.getText(self, "새 폴더", "폴더명:")
if not ok or not name.strip():
return
target = Path(self.current_path, name.strip())
try:
target.mkdir(parents=False, exist_ok=False)
self._load_path(str(target))
except FileExistsError:
QMessageBox.warning(self, "경고", "동일한 이름의 폴더가 이미 존재합니다.")
except PermissionError:
QMessageBox.critical(self, "오류", "폴더를 만들 권한이 없습니다.")
except Exception as e:
QMessageBox.critical(self, "오류", f"폴더 생성 실패: {e}")
# ---------- 뷰 상호작용 ----------
def _item_clicked(self, index: QModelIndex):
path = self.model.filePath(index)
self.path_edit.setText(path)
self._update_ok_enabled()
def _item_double_clicked(self, index: QModelIndex):
path = self.model.filePath(index)
if os.path.isdir(path):
self._load_path(path)
else:
self.path_edit.setText(path)
self._update_ok_enabled()
if self.mode == 'open' and self.select_mode == 'file':
self.accept()
# ---------- 선택 유효성 ----------
def _is_selection_valid(self):
target = self.path_edit.text().strip()
if not target:
return False
if self.select_mode == 'folder':
return os.path.isdir(target)
if self.mode == 'open':
if os.path.isfile(target):
return self._match_type_filters(Path(target).name)
return False
else:
parent = Path(target).parent
return parent.exists() and parent.is_dir()
def _match_type_filters(self, filename: str) -> bool:
text = self.type_combo.currentText()
patterns = []
if "(" in text and ")" in text:
inside = text[text.find("(") + 1:text.find(")")]
patterns = [p.strip() for p in inside.split() if p.strip()]
if not patterns:
return True
for pat in patterns:
if fnmatch.fnmatch(filename, pat):
return True
return False
def _update_ok_enabled(self):
ok_btn = self.button_box.button(QDialogButtonBox.Ok)
ok_btn.setEnabled(self._is_selection_valid())
def _accept_if_valid(self):
if self._is_selection_valid():
self.accept()
else:
QMessageBox.warning(self, "선택 확인", "선택이 유효하지 않습니다.")
# ---------- 경로 입력/자동완성(멀티프로세싱) ----------
def _on_path_edited(self, _):
# 입력 중에는 루트 변경 금지. 자동완성만 트리거.
self._debounce_timer.start()
self._update_ok_enabled()
def _spawn_completer_task(self):
text = self.path_edit.text().strip()
if not text:
self._slm.setStringList([])
return
p = Path(text)
if text.endswith(os.sep):
base_dir = text
prefix = ""
else:
base_dir = str(p.parent) if str(p.parent) else os.sep
prefix = p.name
# 루트/의심 경로에서 짧은 접두사면 미리 차단
sbase = base_dir.lower()
if base_dir in (os.sep, "\\\\") or sbase.startswith("/volumes") or sbase.startswith("/mnt") or sbase.startswith("/net"):
if len(prefix) < 2:
self._slm.setStringList([])
return
want_files = (self.select_mode == 'file')
want_dirs = True
# 이전 요청 취소 시도 (spawn된 프로세스 작업은 즉시 안 꺼질 수 있으나, 우리는 결과를 무시)
if self._pending_future and not self._pending_future.done():
self._pending_future.cancel()
self._task_seq += 1
seq = self._task_seq
# 제출
fut = self._proc_pool.submit(_collect_completions, base_dir, prefix, want_files, want_dirs, 10, 1, 80)
self._pending_future = fut
# 완료 콜백: 워커 스레드에서 실행되므로 Qt로 신호 전달
def _on_done(f):
try:
results = f.result() if f.done() else []
except Exception:
results = []
self._proc_emitter.resultsReady.emit(seq, results)
fut.add_done_callback(_on_done)
@Slot(int, list)
def _on_proc_results(self, seq: int, results: list):
if seq != self._task_seq:
return # 최신 요청만 반영
items = results[:10]
# 이전과 동일하면 갱신 생략
if self._slm.stringList() == items:
return
self._slm.setStringList(items)
# 팝업은 조건부로만 호출
if items and self.path_edit.hasFocus() and self.completer.popup().isVisible():
self.completer.complete()
# ---------- 종료 처리 ----------
def closeEvent(self, e):
try:
if self._pending_future:
self._pending_future.cancel()
self._proc_pool.shutdown(wait=False, cancel_futures=True)
except Exception:
pass
super().closeEvent(e)
# ---------- 외부 API ----------
def selected_path(self) -> str:
return self.path_edit.text().strip()
# ==========================================
# 실행 예시
# ==========================================
def main():
app = QApplication(sys.argv)
# 예시 1: 파일 Open
dlg1 = CFileDialog(
start_path=os.getcwd(),
mode='open',
select_mode='file',
filters=[
"All Files (*)",
"Text Files (*.txt *.md)",
"Images (*.png *.jpg *.jpeg *.bmp *.gif)"
]
)
if dlg1.exec():
print("OPEN FILE:", dlg1.selected_path())
# 예시 2: 폴더 선택
dlg2 = CFileDialog(
start_path=os.getcwd(),
mode='open',
select_mode='folder',
filters=["All Files (*)"]
)
if dlg2.exec():
print("SELECT FOLDER:", dlg2.selected_path())
# 예시 3: 파일 Save
dlg3 = CFileDialog(
start_path=os.getcwd(),
mode='save',
select_mode='file',
filters=["All Files (*)", "Python (*.py)"]
)
if dlg3.exec():
print("SAVE FILE:", dlg3.selected_path())
sys.exit(0)
if __name__ == "__main__":
# 멀티프로세싱 스폰 모드 사용 시 필요하다면 주석 해제
# from multiprocessing import set_start_method
# set_start_method("spawn", force=True)
main()
메모
-
자동완성은 별도 프로세스에서 돌고, GUI는 메인 스레드에서만 Qt 객체를 건드려서 안전해.
-
루트 경로에서 접두사가 짧을 때는 탐색을 생략해 폭주를 막았고,
scandir()에 80ms 타임박스를 걸어 타자 중 지연을 억제했어. -
이전 요청은
cancel()하고seq로 최신성만 보장해 결과 레이스를 제거했어. -
필요하면
time_budget_ms나depth_limit을 조절해서 네트워크 드라이브에서 더 보수적으로 동작하게 만들 수 있어.
특정 OS나 네트워크 공유에서 튜닝이 더 필요하면, 문제 재현 경로와 함께 알려줘. 타임박스·깊이·루트 규칙을 미세조정한 패치를 바로 만들어줄게.
원인을 좀 더 좁힌거 같아. os.path.isfile("/user/j") 이게 너무 느려 안 끝나
맞아, 핵심 원인은 os.path.isfile("/user/j") 같은 “경로 자체에 대한 stat 호출”이 네트워크·자동마운트(autofs)·SMB/UNC 같은 느린 파일시스템에서 블로킹되기 때문이야. 특히 /user 가 자동마운트 지점이면, isfile 한 번만으로도 커널이 마운트를 시도→DNS 탐색→타임아웃까지 기다리느라 GUI가 얼어붙은 것처럼 느껴져. 스레드로 돌려도 GIL·결과 반영 타이밍·연쇄 호출 때문에 체감 프리즈가 나오는 거고.
해법은 “타이핑 중에는 절대 isfile/exists 같은 직접 stat을 하지 않는다”야. 검증이 필요해도 경량 우회로를 쓰거나, 별도 프로세스에서 타임박스로 돌리고 결과가 늦으면 그냥 “모름”으로 처리하는 정책이 안전해.
아래처럼 바꾸면 바로 효과를 볼 수 있어.
전략 요약
-
입력 중에는
isfile완전 금지. 자동완성은 문자열 매칭만. -
최종 OK 검증에서만 “빠른 존재 확인”을 시도하되, 경로에 직접 stat 하지 말고 상위 폴더를 짧게 나열해 이름 비교한다.
-
그래도 네트워크일 수 있으니 타임박스(예: 50ms)를 넘기면 “미확인”으로 간주하고 OK을 잠시 비활성화하거나, 메시지로 알려준다.
-
필요하면 멀티프로세싱으로 “느릴 수 있는 stat”을 분리하고 100~150ms 타임아웃을 둔다.
-
autofs·네트워크 루트 프리픽스를 블록리스트로 두고, 접두사가 짧을 땐 아예 검증을 스킵한다.
코드 패치 1: 빠른 파일 확인기로 교체
_is_selection_valid()에서 쓰던 os.path.isfile()를 아래 함수로 바꿔 호출해.
def _fast_isfile(self, path: str, budget_ms: int = 50) -> bool:
"""경로 자체를 stat하지 않고, 부모 디렉터리를 짧게 나열해서 이름으로 판단."""
p = Path(path)
parent = p.parent
name = p.name
if not name:
return False
# 위험 프리픽스는 즉시 포기(autofs/네트워크 루트)
s = str(parent).lower()
risky_roots = (os.sep, "/net", "/nfs", "/afs", "/mnt", "/user", "/users",
"/volumes", "\\\\")
if any(s.startswith(rr) for rr in risky_roots) and len(name) < 2:
return False # 접두사가 짧으면 스캔하지 않음
t0 = time.perf_counter()
try:
with os.scandir(parent) as it:
for entry in it:
# 타임박스 초과 시 즉시 포기
if (time.perf_counter() - t0) * 1000 > budget_ms:
return False
if entry.name == name:
# 여기서도 stat 대신 dir 여부만 확인(캐시 활용)
try:
return not entry.is_dir(follow_symlinks=False)
except Exception:
return True # 파일로 가정
except Exception:
return False
return False
그리고 _is_selection_valid()를 이렇게 조정.
def _is_selection_valid(self):
target = self.path_edit.text().strip()
if not target:
return False
if self.select_mode == 'folder':
# 폴더 검증도 빠른 버전으로(직접 stat 대신 상위 나열)
return self._fast_isdir(target)
if self.mode == 'open':
# 느린 환경에서 isfile 금지 → 빠른 확인기 사용
return self._fast_isfile(target) and self._match_type_filters(Path(target).name)
else:
parent = Path(target).parent
# 저장은 상위 폴더만 확인(이것도 빠른 판정 사용)
return self._fast_isdir(str(parent))
def _fast_isdir(self, path: str, budget_ms: int = 50) -> bool:
p = Path(path)
parent, name = p.parent, p.name
if not name:
# 루트 등: 대개 디렉터리로 취급
return True
t0 = time.perf_counter()
try:
with os.scandir(parent) as it:
for entry in it:
if (time.perf_counter() - t0) * 1000 > budget_ms:
return False
if entry.name == name:
try:
return entry.is_dir(follow_symlinks=False)
except Exception:
return True
except Exception:
return False
return False
설명: os.path.isfile(target) 대신, 부모 디렉터리를 아주 짧게 나열하면서 이름이 같은 엔트리만 판정한다. 이러면 커널이 문제의 경로 자체를 접근(=자동마운트 트리거)하지 않고도 대부분의 케이스를 해결한다.
코드 패치 2: 최종 검증을 비동기·타임아웃으로
그래도 원격 부모 폴더 나열이 느릴 수 있어. 그럴 땐 프로세스 풀로 “안전 판정”을 돌리고, 120ms 안에 답이 없으면 OK 버튼을 잠깐 비활성화하거나 경고 배지를 띄우자.
최상위에 워커 함수 추가:
def _safe_isfile_worker(path: str) -> bool:
try:
return os.path.isfile(path)
except Exception:
return False
다이얼로그 __init__에서 재사용용 풀 하나:
from concurrent.futures import ProcessPoolExecutor
self._check_pool = ProcessPoolExecutor(max_workers=1)
self._pending_check = None
OK 직전 확인(예: _accept_if_valid에서):
def _accept_if_valid(self):
target = self.path_edit.text().strip()
if self.mode == 'open' and self.select_mode == 'file':
# 빠른 판정으로 충분하면 바로 통과
if self._fast_isfile(target) and self._match_type_filters(Path(target).name):
self.accept(); return
# 느리면 프로세스로 120ms만 시도
if self._pending_check and not self._pending_check.done():
self._pending_check.cancel()
fut = self._check_pool.submit(_safe_isfile_worker, target)
self._pending_check = fut
try:
ok = fut.result(timeout=0.12)
except Exception:
ok = False
if ok and self._match_type_filters(Path(target).name):
self.accept()
return
QMessageBox.warning(self, "선택 확인", "파일이 존재하지 않거나 접근할 수 없습니다.")
return
# 나머지 케이스는 기존 로직
if self._is_selection_valid():
self.accept()
else:
QMessageBox.warning(self, "선택 확인", "선택이 유효하지 않습니다.")
닫을 때 풀 정리:
def closeEvent(self, e):
try:
if self._pending_check: self._pending_check.cancel()
self._check_pool.shutdown(wait=False, cancel_futures=True)
except Exception:
pass
super().closeEvent(e)
정책 팁
-
“루트/네트워크 루트 + 접두사 짧음”에서는 어떤 판정도 하지 말고, 자동완성과 OK 둘 다 보수적으로 막아라. 타자 몇 글자 더 치면 자연스럽게 풀린다.
-
저장 모드에서는 존재 확인이 필요 없다. 상위 폴더만 빠르게 확인하면 된다.
-
자동완성·검증 모두 “경로 자체 stat”을 피하고, 가능하면 “부모 나열”로 판정하라.
-
조직 환경에서
/user,/net,/home,/nfs같은 autofs 지점이 있다면, 그 접두사는 공격적으로 스킵하도록 블록리스트에 올려라. -
리눅스라면
/proc/mounts를 읽어autofs타입 마운트 목록을 만들어두고, 그 경로 밑은 접두사 길이 3 미만이면 무조건 스킵하는 게 최선이다.
정리하면, 지금 보이는 병목은 isfile 한 방으로 autofs를 찌르면서 발생해. 위 패치처럼 “경로 자체 stat 금지 + 부모 디렉터리 짧게 나열 + 타임박스 + 필요시 프로세스 타임아웃”으로 바꾸면 같은 환경에서도 프리즈가 사라질 거야. 원하는 형태로 바로 붙일 수 있게, 네 코드의 _is_selection_valid() 부분을 기반으로 한 미니 패치도 만들어줄 수 있어.
음 오케이오케이. 결국 내가 원하는건 내부에서 어떻게 동작하든 깔끔하게 isfile 같은 함수를 다 구현해주면 돼. 파일 확인 함수 모두 구현해줘
좋아. 내부 구현이 어떻게 돌아가든—UI는 몰라도 되도록—파일 확인용 유틸을 한 덩어리로 정리해줄게. 핵심 목표는 다음이야.
-
타이핑 중엔 느린 stat 계열 호출을 피한다.
-
부모 디렉터리 나열로 빠르게 추정한다(타임박스).
-
필요하면 멀티프로세싱으로 짧은 타임아웃 내 “확정” 검증을 한다.
-
호출자가 정치(빠름/정확)만 고르면 된다.
아래 모듈은 바로 붙여서 쓸 수 있어. 반환 정책은 세 가지로 나눈다.
-
빠른 추정 계열(
quick_…): True/False/None(None=시간 초과/판단 불가) -
확정 검증 계열(
confirm_…): 멀티프로세싱 + 타임아웃으로 True/False -
안전 래퍼(
safe_…): quick이 결정 못 하면 confirm을 짧게 시도
# file_checks.py
# -*- coding: utf-8 -*-
import os
import time
from pathlib import Path
from typing import Optional, Tuple
from concurrent.futures import ProcessPoolExecutor, TimeoutError
# ---------------------------------------
# 내부 설정
# ---------------------------------------
# autofs/네트워크 루트 등 "건드리면 느릴" 수 있는 접두사
_RISKY_PREFIXES = (
os.sep, # "/"
"/net", "/nfs", "/afs", "/mnt", "/media",
"/user", "/users", "/home",
"/Volumes", # macOS
"\\\\", # Windows UNC
)
# Windows 드라이브 루트 후보 (필요시 추가)
_WIN_ROOTS = tuple(f"{d}:\\" for d in "ABCDEFGHIJKLMNOPQRSTUVWXYZ")
def _is_risky_root(path_str: str) -> bool:
s = path_str.replace("\\", "/").lower()
return any(s.startswith(p.lower()) for p in _RISKY_PREFIXES) or path_str in _WIN_ROOTS
# ---------------------------------------
# 빠른 부모-나열 기반 판정(타임박스)
# ---------------------------------------
def _parent_scan_type(path: str, budget_ms: int, follow_symlinks: bool) -> Optional[str]:
"""
부모 디렉터리를 os.scandir로 짧게 훑어 파일/디렉터리/기타 여부를 추정한다.
반환: "file" | "dir" | "other" | None(판단불가/시간초과)
"""
p = Path(path)
parent, name = p.parent, p.name
if not name:
# 루트 같은 케이스: 디렉터리 취급
return "dir"
t0 = time.perf_counter()
try:
with os.scandir(parent) as it:
for entry in it:
if (time.perf_counter() - t0) * 1000 > budget_ms:
return None
if entry.name != name:
continue
try:
if entry.is_dir(follow_symlinks=follow_symlinks):
return "dir"
elif entry.is_file(follow_symlinks=follow_symlinks):
return "file"
else:
return "other"
except Exception:
# entry 질의 실패 시 파일로 가정할 수도 있지만 보수적으로 기타 처리
return "other"
except Exception:
return None
return None
def quick_isfile(path: str, budget_ms: int = 50) -> Optional[bool]:
"""
빠른 추정: 부모 나열만으로 파일 여부를 판단.
True/False 또는 None(시간초과/판단불가) 반환.
"""
typ = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (typ == "file") if typ is not None else None
def quick_isdir(path: str, budget_ms: int = 50) -> Optional[bool]:
typ = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (typ == "dir") if typ is not None else None
def quick_exists(path: str, budget_ms: int = 50) -> Optional[bool]:
typ = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (typ in ("file", "dir", "other")) if typ is not None else None
# ---------------------------------------
# 멀티프로세싱 확정 검증(타임아웃)
# ---------------------------------------
def _mp_isfile(path: str) -> bool:
try:
return os.path.isfile(path)
except Exception:
return False
def _mp_isdir(path: str) -> bool:
try:
return os.path.isdir(path)
except Exception:
return False
def _mp_exists(path: str) -> bool:
try:
return os.path.exists(path)
except Exception:
return False
def _confirm_with_timeout(fn, path: str, timeout_ms: int) -> bool:
# 프로세스를 매 호출마다 만드는 방식: 누수 없음, 간결
timeout_s = max(0, timeout_ms) / 1000.0
with ProcessPoolExecutor(max_workers=1) as pool:
fut = pool.submit(fn, path)
try:
return bool(fut.result(timeout=timeout_s))
except TimeoutError:
return False
except Exception:
return False
def confirm_isfile(path: str, timeout_ms: int = 150) -> bool:
return _confirm_with_timeout(_mp_isfile, path, timeout_ms)
def confirm_isdir(path: str, timeout_ms: int = 150) -> bool:
return _confirm_with_timeout(_mp_isdir, path, timeout_ms)
def confirm_exists(path: str, timeout_ms: int = 150) -> bool:
return _confirm_with_timeout(_mp_exists, path, timeout_ms)
# ---------------------------------------
# 안전 래퍼: 빠름→(필요 시)확정
# ---------------------------------------
def safe_isfile(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150,
risky_skip_minlen: int = 2) -> bool:
"""
권장 사용: 타자 중엔 빠르게, 확정이 필요하면 짧게 confirm.
- 위험 루트 + 이름이 짧으면(기본 2) 바로 False로 간주(탐색 회피).
"""
p = Path(path)
if _is_risky_root(str(p.parent)) and len(p.name) < risky_skip_minlen:
return False
q = quick_isfile(path, budget_ms=quick_budget_ms)
if q is not None:
return q
return confirm_isfile(path, timeout_ms=confirm_timeout_ms)
def safe_isdir(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150,
risky_skip_minlen: int = 2) -> bool:
p = Path(path)
if _is_risky_root(str(p.parent)) and len(p.name) < risky_skip_minlen:
# 루트 자체는 디렉터리 취급
return str(p.name) == ""
q = quick_isdir(path, budget_ms=quick_budget_ms)
if q is not None:
return q
return confirm_isdir(path, timeout_ms=confirm_timeout_ms)
def safe_exists(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150,
risky_skip_minlen: int = 2) -> bool:
p = Path(path)
if _is_risky_root(str(p.parent)) and len(p.name) < risky_skip_minlen:
return False
q = quick_exists(path, budget_ms=quick_budget_ms)
if q is not None:
return q
return confirm_exists(path, timeout_ms=confirm_timeout_ms)
# ---------------------------------------
# 접근/권한 체크(빠른 버전)
# ---------------------------------------
def quick_access(path: str, mode: int = os.R_OK, budget_ms: int = 50) -> Optional[bool]:
"""
접근 권한 추정. 존재 여부를 확정하지 않고 부모 나열로 보수 판정.
파일이면 os.R_OK, os.W_OK, os.X_OK 의미 동일.
"""
typ = _parent_scan_type(path, budget_ms, follow_symlinks=False)
if typ is None:
return None
# 존재한다고 가정하고 접근 모드는 낙관적으로 추정할 수도 있지만,
# 네트워크/권한 이슈는 확인 비용이 큼 → None 반환이 안전
return None
def confirm_access(path: str, mode: int = os.R_OK, timeout_ms: int = 150) -> bool:
def _mp_access(p: str, m: int) -> bool:
try:
return os.access(p, m)
except Exception:
return False
timeout_s = max(0, timeout_ms) / 1000.0
with ProcessPoolExecutor(max_workers=1) as pool:
fut = pool.submit(_mp_access, path, mode)
try:
return bool(fut.result(timeout=timeout_s))
except TimeoutError:
return False
except Exception:
return False
def safe_access(path: str, mode: int = os.R_OK,
quick_budget_ms: int = 50, confirm_timeout_ms: int = 150) -> bool:
q = quick_exists(path, budget_ms=quick_budget_ms)
if q is False:
return False
# q is True/None → 필요 시 짧게 확인
return confirm_access(path, mode=mode, timeout_ms=confirm_timeout_ms)
# ---------------------------------------
# 편의: 파일/폴더 모드에서 OK 활성화 판정
# ---------------------------------------
def can_accept_open_file(path: str,
require_filter_match: bool = False,
name_patterns: Tuple[str, ...] = (),
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
"""
Open+File 모드의 OK 가능 여부.
"""
if not path:
return False
if not safe_isfile(path, quick_budget_ms, confirm_timeout_ms):
return False
if require_filter_match and name_patterns:
import fnmatch
name = Path(path).name
return any(fnmatch.fnmatch(name, pat) for pat in name_patterns)
return True
def can_accept_open_folder(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
if not path:
return False
return safe_isdir(path, quick_budget_ms, confirm_timeout_ms)
def can_accept_save_file(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
"""
Save+File: 상위 폴더만 존재하면 OK.
"""
if not path:
return False
parent = str(Path(path).parent)
return safe_isdir(parent, quick_budget_ms, confirm_timeout_ms)
간단 사용 예시는 아래처럼.
# 예시
from file_checks import safe_isfile, safe_isdir, safe_exists, can_accept_open_file
print(safe_isfile("/user/j")) # autofs 환경에서도 프리즈 없이 빠른 반환
print(safe_isdir("/some/dir")) # 빠른 추정 → 필요시 150ms 타임아웃으로 확정
print(safe_exists("/network/share/file.txt")) # 동일 정책
print(can_accept_open_file("/path/a.png",
require_filter_match=True,
name_patterns=("*.png","*.jpg")))
핵심 포인트 정리
-
quick 단계는 부모 디렉터리만 짧게 나열해서 추정하므로 autofs/UNC에서도 안전하다.
-
confirm 단계는 별도 프로세스로
os.path.isfile같은 “진짜 stat”을 돌리되, 타임아웃을 걸어 GUI가 멈추지 않게 한다. -
safe_*는 quick→confirm 순으로 처리해서 UI 코드가 간단해진다. -
위험 루트에서 이름이 너무 짧으면(기본 2) 아예 건드리지 않고 False를 돌려 탐욕적 마운트를 피한다. 필요하면 이 기준을 조정하면 된다.
원하면 이 모듈을 너의 CFileDialog에 바로 끼워 넣을 수 있도록 _is_selection_valid() 패치 버전도 만들어서 붙여줄게.
risky root를 하드코딩으로 하는건 별로 안하고 싶네. 그 부분은 빼줘
좋아. 하드코딩된 “risky root” 판단을 싹 제거한 버전이야.
정책은 오직 시간 예산(타임박스)과 멀티프로세싱 타임아웃으로만 안전성을 보장해. 필요하면 호출 시 파라미터로 시간 값을 조절해 쓰면 된다.
# file_checks.py
# -*- coding: utf-8 -*-
import os
import time
from pathlib import Path
from typing import Optional, Tuple
from concurrent.futures import ProcessPoolExecutor, TimeoutError
# ---------------------------------------
# 부모-나열 기반 빠른 추정(타임박스)
# ---------------------------------------
def _parent_scan_type(path: str, budget_ms: int, follow_symlinks: bool) -> Optional[str]:
"""
부모 디렉터리를 짧게 나열해 대상이 파일/디렉터리/기타인지 추정.
반환: "file" | "dir" | "other" | None(시간초과/판단불가)
"""
p = Path(path)
parent, name = p.parent, p.name
if not name:
# 루트 같은 케이스는 디렉터리로 간주
return "dir"
t0 = time.perf_counter()
try:
with os.scandir(parent) as it:
for entry in it:
if (time.perf_counter() - t0) * 1000 > budget_ms:
return None
if entry.name != name:
continue
try:
if entry.is_dir(follow_symlinks=follow_symlinks):
return "dir"
elif entry.is_file(follow_symlinks=follow_symlinks):
return "file"
else:
return "other"
except Exception:
return "other"
except Exception:
return None
return None
def quick_isfile(path: str, budget_ms: int = 50) -> Optional[bool]:
typ = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (typ == "file") if typ is not None else None
def quick_isdir(path: str, budget_ms: int = 50) -> Optional[bool]:
typ = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (typ == "dir") if typ is not None else None
def quick_exists(path: str, budget_ms: int = 50) -> Optional[bool]:
typ = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (typ in ("file", "dir", "other")) if typ is not None else None
# ---------------------------------------
# 멀티프로세싱 확정 검증(타임아웃)
# ---------------------------------------
def _mp_isfile(path: str) -> bool:
try:
return os.path.isfile(path)
except Exception:
return False
def _mp_isdir(path: str) -> bool:
try:
return os.path.isdir(path)
except Exception:
return False
def _mp_exists(path: str) -> bool:
try:
return os.path.exists(path)
except Exception:
return False
def _confirm_with_timeout(fn, path: str, timeout_ms: int) -> bool:
timeout_s = max(0, timeout_ms) / 1000.0
# 호출 단위 풀 생성: 간단하고 누수 없음
with ProcessPoolExecutor(max_workers=1) as pool:
fut = pool.submit(fn, path)
try:
return bool(fut.result(timeout=timeout_s))
except TimeoutError:
return False
except Exception:
return False
def confirm_isfile(path: str, timeout_ms: int = 150) -> bool:
return _confirm_with_timeout(_mp_isfile, path, timeout_ms)
def confirm_isdir(path: str, timeout_ms: int = 150) -> bool:
return _confirm_with_timeout(_mp_isdir, path, timeout_ms)
def confirm_exists(path: str, timeout_ms: int = 150) -> bool:
return _confirm_with_timeout(_mp_exists, path, timeout_ms)
# ---------------------------------------
# 안전 래퍼: 빠름→(필요 시)확정
# ---------------------------------------
def safe_isfile(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
"""
1) quick(부모 나열, 타임박스)로 결정 시 그 값 사용
2) 결정 불가(None)이면 멀티프로세싱으로 짧게 확정
"""
q = quick_isfile(path, budget_ms=quick_budget_ms)
return q if q is not None else confirm_isfile(path, timeout_ms=confirm_timeout_ms)
def safe_isdir(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
q = quick_isdir(path, budget_ms=quick_budget_ms)
return q if q is not None else confirm_isdir(path, timeout_ms=confirm_timeout_ms)
def safe_exists(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
q = quick_exists(path, budget_ms=quick_budget_ms)
return q if q is not None else confirm_exists(path, timeout_ms=confirm_timeout_ms)
# ---------------------------------------
# 접근/권한 체크
# ---------------------------------------
def confirm_access(path: str, mode: int = os.R_OK, timeout_ms: int = 150) -> bool:
def _mp_access(p: str, m: int) -> bool:
try:
return os.access(p, m)
except Exception:
return False
timeout_s = max(0, timeout_ms) / 1000.0
with ProcessPoolExecutor(max_workers=1) as pool:
fut = pool.submit(_mp_access, path, mode)
try:
return bool(fut.result(timeout=timeout_s))
except TimeoutError:
return False
except Exception:
return False
def safe_access(path: str, mode: int = os.R_OK,
quick_budget_ms: int = 50, confirm_timeout_ms: int = 150) -> bool:
"""
접근성은 빠른 추정이 의미가 적으므로 바로 확정 호출을 권장.
그래도 호출 시그니처를 맞추기 위해 quick_budget_ms를 남겨둠.
"""
return confirm_access(path, mode=mode, timeout_ms=confirm_timeout_ms)
# ---------------------------------------
# OK 활성화 판정(다이얼로그 용)
# ---------------------------------------
def can_accept_open_file(path: str,
require_filter_match: bool = False,
name_patterns: Tuple[str, ...] = (),
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
if not path:
return False
if not safe_isfile(path, quick_budget_ms, confirm_timeout_ms):
return False
if require_filter_match and name_patterns:
import fnmatch
name = Path(path).name
return any(fnmatch.fnmatch(name, pat) for pat in name_patterns)
return True
def can_accept_open_folder(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
if not path:
return False
return safe_isdir(path, quick_budget_ms, confirm_timeout_ms)
def can_accept_save_file(path: str,
quick_budget_ms: int = 50,
confirm_timeout_ms: int = 150) -> bool:
if not path:
return False
parent = str(Path(path).parent)
return safe_isdir(parent, quick_budget_ms, confirm_timeout_ms)
포인트
-
어떤 경로든 하드코딩 규칙 없이 동일하게 다룬다.
-
quick 계열은 부모 디렉터리만 짧게 스캔해서 프리즈를 막고, 판단 불가일 때만 confirm(프로세스, 타임아웃)으로 확정한다.
-
UI에서는
safe_isfile/safe_isdir/safe_exists만 호출하면 되고, 느려 보이면quick_budget_ms를 줄이거나confirm_timeout_ms를 늘려 튜닝하면 된다.
원하면 이 모듈을 CFileDialog의 _is_selection_valid()에 바로 붙인 예시도 덧붙여줄게.
worker가 1개인데 만약에 async나 thread로 confirm_isfile이 불려도 상관없나?
결론부터 말하면, 지금 올려둔 confirm_isfile()처럼 호출할 때마다ProcessPoolExecutor(max_workers=1)를 만들고 닫는 구조라면,
여러 스레드·async 컨텍스트에서 동시에 호출되어도 서로 독립적으로 돌아서 문제 없습니다. 각 호출이 자기 전용 프로세스를 잠깐 띄웠다가 타임아웃/완료 후 종료하니까요. 다만 동시 호출 수만큼 프로세스 생성 오버헤드 가 생깁니다(특히 Windows/macOS에서 spawn 비용 큼).
원하는 운영 방식에 따라 선택지가 두 가지예요.
옵션 A: 지금처럼 “호출 단위 풀 생성”
-
장점: 동시 호출 안전(완전 독립), 리소스 누수 없음, 구현 간단.
-
단점: 호출이 잦거나 동시에 많이 들어오면 프로세스 생성 오버헤드↑.
-
스레드/async에서 동시에 불러도 OK. 단, async에서는 반드시 이벤트루프를 블로킹하지 않게 감싸야 함(아래 참고).
옵션 B: “공용 풀”을 하나만 두고 직렬/병렬 제어
-
장점: 프로세스 생성/해제 오버헤드↓, 캐시를 얹기도 쉬움.
-
단점:
max_workers=1로 두면 요청이 큐에 쌓여 직렬 처리 됩니다. 여러 호출이 동시에 들어오면 뒤에 오는 건 타임아웃될 수 있어요. 병렬을 원하면 2~4로 늘리면 됩니다.
아래는 두 옵션의 안전한 패턴과 async/스레드에서 쓰는 방법입니다.
옵션 A 유지(호출 단위 풀) + async/스레드에서 호출
# 동기 호출은 그대로 사용
ok = confirm_isfile(path, timeout_ms=150)
# 스레드에서 호출 (GUI 이벤트 루프 안막음)
import threading
t = threading.Thread(target=lambda: confirm_isfile(path), daemon=True)
t.start()
# asyncio에서 호출 (반드시 run_in_executor/to_thread로 감싸기)
import asyncio
async def confirm_isfile_async(path, timeout_ms=150):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, confirm_isfile, path, timeout_ms)
# 사용
ok = await confirm_isfile_async(path)
옵션 B: 공용 풀(권장 파라미터 예시: max_workers=2~4)
# file_checks_pool.py
import os
from concurrent.futures import ProcessPoolExecutor, TimeoutError
_pool = ProcessPoolExecutor(max_workers=2) # 병렬 2개
def _mp_isfile(path:str)->bool:
try: return os.path.isfile(path)
except: return False
def confirm_isfile(path:str, timeout_ms:int=150)->bool:
fut = _pool.submit(_mp_isfile, path)
try:
return bool(fut.result(timeout=timeout_ms/1000))
except TimeoutError:
return False
except Exception:
return False
# 종료 시 정리(앱 종료 훅이나 다이얼로그 closeEvent에서 호출)
def shutdown_pool():
try: _pool.shutdown(wait=False, cancel_futures=True)
except: pass
async 래퍼
# asyncio에서 블로킹 피하기
import asyncio
from file_checks_pool import confirm_isfile as _confirm_isfile
async def confirm_isfile_async(path, timeout_ms=150):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, _confirm_isfile, path, timeout_ms)
스레드 안전성
-
ProcessPoolExecutor자체는 스레드 세이프 한 제출을 지원합니다. 여러 스레드에서 동시에submit()호출 가능. -
단,
max_workers=1이면 제출은 되지만 실행은 순차적 입니다. 여러 곳에서 동시에 타임아웃 짧게 걸고 기다리면, 뒤의 Future들이 타임아웃으로 False를 받게 되는 건 정상 동작입니다. 필요하면max_workers를 늘리거나, 호출 측에서 디바운싱/리밋을 두세요.
추가 팁
-
quick_*(부모 나열 기반 추정)는 순수 I/O + 지역 루프이므로 스레드·async 동시 호출 안전 합니다. 공유 상태가 없고, GIL로 간단히 보호됩니다. -
Qt/async 혼합 환경이면, 이벤트 루프를 막지 않도록
confirm_*를 항상run_in_executor(또는to_thread)로 감싸세요. -
Windows/macOS에서 멀티프로세싱은 항상
if __name__ == "__main__":가드가 필요합니다. Qt 앱 엔트리도 그 안에서 시작하세요. -
공용 풀을 쓴다면 앱 종료 시
shutdown_pool()호출로 정리하는 습관을 들이면 좋습니다.
정리:
-
지금 구조(호출 단위 풀)는 동시 호출에도 안전 하지만 오버헤드가 큽니다.
-
공용 풀은 효율적 이고 스레드·async 모두 안전하되,
max_workers설정에 따라 직렬/병렬 특성이 달라지니 호출 패턴에 맞춰 수치만 잡으면 됩니다.
근데 보니까 aiofiles 라는 패키지가 있네. 멀티프로세싱 대신 이걸 타임아웃 걸어서 쓰는건 어떨까
좋은 발상인데, 딱 정리하면 이렇습니다.
-
aiofiles는 파일 열기/읽기 같은 파일 I/O를 비동기로 다루기 위해 내부적으로 스레드풀을 씁니다.
os.path.isfile같은 stat 계열은 aiofiles가 감싸주지 않으니asyncio.to_thread나run_in_executor로 직접 감싸야 합니다. -
타임아웃은
asyncio.wait_for로 손쉽게 걸 수 있지만, 커널에서 stat이 오래 블록되면 스레드는 계속 붙잡혀 있고 코루틴만 취소됩니다. 즉, GUI는 살지만 백그라운드 스레드가 “끝날 때까지” 남습니다. 반면 멀티프로세싱은 프로세스를 내려서 정리할 수 있다는 차이가 있어요. -
정리: 이미 앱에 asyncio 루프(예: qasync 사용)가 있고, 네트워크 환경이 아주 험하지 않다면 aiofiles/스레드 경로가 간단하고 충분합니다. 오랫동안 먹통이 되는 경로가 많다면 멀티프로세싱이 더 안전합니다.
아래는 깔끔한 비동기 버전입니다. 하드코딩 규칙 없이, 타임박스 중심으로 동작합니다.
1) 빠른 추정(동기, 부모 나열 타임박스)
import os, time
from pathlib import Path
from typing import Optional
def _parent_scan_type(path: str, budget_ms: int, follow_symlinks: bool) -> Optional[str]:
p = Path(path); parent, name = p.parent, p.name
if not name:
return "dir"
t0 = time.perf_counter()
try:
with os.scandir(parent) as it:
for entry in it:
if (time.perf_counter() - t0) * 1000 > budget_ms:
return None
if entry.name != name:
continue
try:
if entry.is_dir(follow_symlinks=follow_symlinks):
return "dir"
elif entry.is_file(follow_symlinks=follow_symlinks):
return "file"
else:
return "other"
except Exception:
return "other"
except Exception:
return None
return None
def quick_isfile(path: str, budget_ms: int = 50) -> Optional[bool]:
t = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (t == "file") if t is not None else None
def quick_isdir(path: str, budget_ms: int = 50) -> Optional[bool]:
t = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (t == "dir") if t is not None else None
def quick_exists(path: str, budget_ms: int = 50) -> Optional[bool]:
t = _parent_scan_type(path, budget_ms, follow_symlinks=False)
return (t in ("file","dir","other")) if t is not None else None
2) aiofiles 없이 asyncio로 “확정 확인” (스레드풀 + 타임아웃)
import os, asyncio
async def confirm_isfile_async(path: str, timeout_ms: int = 150) -> bool:
loop = asyncio.get_running_loop()
coro = loop.run_in_executor(None, os.path.isfile, path)
try:
return await asyncio.wait_for(coro, timeout=timeout_ms/1000)
except asyncio.TimeoutError:
return False
except Exception:
return False
async def confirm_isdir_async(path: str, timeout_ms: int = 150) -> bool:
loop = asyncio.get_running_loop()
coro = loop.run_in_executor(None, os.path.isdir, path)
try:
return await asyncio.wait_for(coro, timeout=timeout_ms/1000)
except asyncio.TimeoutError:
return False
except Exception:
return False
async def confirm_exists_async(path: str, timeout_ms: int = 150) -> bool:
loop = asyncio.get_running_loop()
coro = loop.run_in_executor(None, os.path.exists, path)
try:
return await asyncio.wait_for(coro, timeout=timeout_ms/1000)
except asyncio.TimeoutError:
return False
except Exception:
return False
참고: aiofiles를 꼭 쓰고 싶다면 await asyncio.wait_for(asyncio.to_thread(os.path.isfile, path), timeout)처럼 동일하게 감싸면 됩니다. 실질적으로는 위 코드와 같습니다.
3) 안전 래퍼: 빠른 추정 → 필요 시 비동기 확정
async def safe_isfile_async(path: str, quick_budget_ms=50, confirm_timeout_ms=150) -> bool:
q = quick_isfile(path, budget_ms=quick_budget_ms)
return q if q is not None else await confirm_isfile_async(path, timeout_ms=confirm_timeout_ms)
async def safe_isdir_async(path: str, quick_budget_ms=50, confirm_timeout_ms=150) -> bool:
q = quick_isdir(path, budget_ms=quick_budget_ms)
return q if q is not None else await confirm_isdir_async(path, timeout_ms=confirm_timeout_ms)
async def safe_exists_async(path: str, quick_budget_ms=50, confirm_timeout_ms=150) -> bool:
q = quick_exists(path, budget_ms=quick_budget_ms)
return q if q is not None else await confirm_exists_async(path, timeout_ms=confirm_timeout_ms)
4) PySide6와 같이 쓰는 방법
-
Qt 이벤트 루프와 asyncio를 함께 쓰려면 qasync를 쓰는 게 깔끔합니다.
-
버튼 클릭 시
asyncio.create_task(safe_isfile_async(...))로 던지고, 완료되면 OK 버튼 상태를 갱신합니다. -
타임아웃이 발생해도 스레드는 죽지 않고 백그라운드에서 끝까지 돌 수 있습니다. 보통 문제 없지만, 호출 폭주를 막으려면 디바운스/동시 실행 제한을 두세요.
5) 멀티프로세싱과 비교한 장단점 요약
-
aiofiles/스레드 방식
-
장점: 설정 간단, 프로세스 스폰 오버헤드 없음, UI와 자연스러운 연동.
-
단점: 타임아웃이 걸려도 실제 OS 호출 스레드는 살아 있어 누적될 수 있음. 아주 느린 네트워크/autofs에서 많이 호출하면 스레드풀 고갈 가능성.
-
-
멀티프로세싱
-
장점: 최악의 블로킹에서 격리 안전. 필요시 프로세스 종료로 강제 정리 가능.
-
단점: 스폰 비용, 코드·배포 복잡도 증가.
-
실전 추천은 이렇습니다.
-
앱이 이미 qasync로 asyncio를 쓰고 있고, 경로 확인 호출 빈도가 낮거나 중간 정도라면 위 비동기 스레드풀+타임아웃 구성이 가장 단순합니다.
-
네트워크 마운트가 “자주” 느려지는 환경이면, 비동기 버전을 기본으로 쓰되, 타임아웃이 일정 횟수 연속 발생할 때만 멀티프로세싱 경로로 자동 폴백하는 하이브리드가 제일 튼튼합니다.