[Python] .style.yapf YAPF (feat. Python formatter)

yapf

YAPF is a Python formatter based on clang-format (developed by Daniel Jasper).

아무리해도 기본은 마음에 안들어서 조금씩 변경해서 사용하고 싶었다. yapf 가 좋다길래 해봤다.

설정한 값.

해당값으로 할때 sqlalchemy로 query를 작성하면 너무 이상하게 줄바꿈이 되어서 전처리 및 주석처리를 해주고있다.

.style.yapf

[style]
based_on_style = pep8
indent_width = 2
column_limit = 200
space_between_ending_comma_and_closing_bracket = false

# enter only trailing comma
SPLIT_ARGUMENTS_WHEN_COMMA_TERMINATED = true
SPLIT_ALL_COMMA_SEPARATED_VALUES = false
SPLIT_BEFORE_FIRST_ARGUMENT = true
# SPLIT_BEFORE_NAMED_ASSIGNS = false

# enter
EACH_DICT_ENTRY_ON_SEPARATE_LINE = true
ALLOW_SPLIT_BEFORE_DICT_VALUE = true
COALESCE_BRACKETS = false
DEDENT_CLOSING_BRACKETS = true
SPLIT_BEFORE_LOGICAL_OPERATOR = true

sqlalchemy 용

# format_sqlalchamy.py
import re
from pathlib import Path

# 후행 연산자 감지
OPERATOR_PATTERN = re.compile(r"\)\s*(or|and|\+|\-|\*|/|if|is|not|in)\b.*")

# 대입문 감지
ASSIGN_PATTERN = re.compile(r"^\s*\w[\w\d_]*\s*=")


def should_refactor(block: str) -> bool:
  return (block.count(".filter") >= 2 or ".join" in block) and len(block.strip()) > 100


def extract_query_block(lines, start_index):
  block = []
  follow = ""
  parens = 0
  i = start_index
  while i < len(lines):
    line = lines[i]
    if i > start_index and ASSIGN_PATTERN.match(line):  # 다음 대입문이 나오면 stop
      break
    block.append(line)
    parens += line.count("(") - line.count(")")
    i += 1
    if parens <= 0 and line.strip().endswith(")"):
      break
  if i < len(lines) and lines[i].strip().startswith("."):
    follow = lines[i].strip()
    i += 1
  return block, follow, i


def split_final_operator(block_lines):
  last_line = block_lines[-1]
  match = OPERATOR_PATTERN.search(last_line)
  if match:
    split_pos = match.start()
    return block_lines[:-1] + [last_line[:split_pos].rstrip()], last_line[split_pos:].strip()
  return block_lines, ""


def reformat_block(assign_indent: str, var_name: str, query_block: str) -> list[str]:
  lines = query_block.strip().split("\n")
  if len(query_block.strip()) <= 100:
    return [f"{assign_indent}{var_name} = {query_block.strip()}"]

  first_line = lines[0].strip()
  base_line = f"{assign_indent}{var_name} = ({first_line}"
  opening_paren_col = base_line.find("(")
  dot_index = base_line.find(".") if "." in base_line else opening_paren_col + 2
  chain_indent = " " * dot_index
  closing_indent = " " * opening_paren_col

  body_lines, trailing_op = split_final_operator(lines[1:])
  rest_lines = [f"{chain_indent}{line.strip()}" for line in body_lines]
  closing = f"{closing_indent})"
  if trailing_op:
    closing += f" {trailing_op}"
  return [base_line, *rest_lines, closing]


def process_code(code: str) -> str:
  lines = code.split("\n")
  output = []
  i = 0
  while i < len(lines):
    line = lines[i]
    if "db.query" in line:
      assign_match = re.match(r"^(\s*)(\w[\w\d_]*)\s*=\s*(db\.query\(.*)", line)
      if assign_match:
        assign_indent = assign_match.group(1)
        var_name = assign_match.group(2)
        query_start = assign_match.group(3)
        lines[i] = assign_indent + query_start
        block, follow, next_i = extract_query_block(lines, i)
        full_query = "\n".join(block)

        if not should_refactor(full_query):
          output.append(f"{assign_indent}{var_name} = {full_query.strip()}")
          i = next_i
          continue

        formatted = re.sub(r"\)\.(filter|join|order_by|limit)", r")\n" + assign_indent + r".\1", full_query)

        output.append(f"{assign_indent}# yapf: disable")
        output.extend(reformat_block(assign_indent, var_name, formatted))
        output.append(f"{assign_indent}# yapf: enable")
        if follow:
          output.append(f"{assign_indent}{follow}")
        i = next_i
        continue
    output.append(line)
    i += 1
  return "\n".join(output)


def run_on_target_dir(target_dir: str, in_place: bool = False, only_inside="databases"):
  base = Path(target_dir)
  py_files = [p for p in base.rglob("*.py") if only_inside in p.parts]
  print(f"\n📁 '{only_inside}/' 폴더 내 .py 파일 {len(py_files)}개 처리 중...\n")
  for path in py_files:
    try:
      original = path.read_text(encoding="utf-8")
      updated = process_code(original)
      if updated != original:
        if in_place:
          path.write_text(updated, encoding="utf-8")
          print(f"✅ 수정됨: {path}")
        else:
          new_path = path.with_name(path.stem + "_yapf_disabled" + path.suffix)
          new_path.write_text(updated, encoding="utf-8")
          print(f"💾 저장됨: {new_path}")
      else:
        print(f"⏭ 변경 없음: {path}")
    except Exception as e:
      print(f"❌ 오류: {path} → {e}")


if __name__ == "__main__":
  import sys
  if len(sys.argv) < 2:
    print("❗ 사용법: python fix_queries.py <project_dir> [--in-place]")
  else:
    run_on_target_dir(target_dir=sys.argv[1], in_place="--in-place" in sys.argv)