Python SQLite 조회: 따옴표 든 코드를 매개변수로 정확히 찾기

설명용 합성 자료의 도해이며 실제 앱 실행 화면이 아닙니다. 세 조회는 각각 별개로 수행합니다. 조회 값은 SQL 구문에 붙이지 않고 한 개의 매개변수로 전달합니다.

상품 코드가 O'REILLY라면 작은따옴표도 코드의 일부입니다. ' OR 1=1 --처럼 SQL 조건으로 보이는 문자열도 등록된 코드라면 그 한 행을 찾아야 합니다. 이번 실습에서는 SQL 문장을 고정하고 검색 값만 매개변수로 전달해, 각 코드가 자기 행 한 개만 반환하는지 확인합니다.

Python 표준 라이브러리만 사용합니다. 가상 CSV를 매번 새 메모리 SQLite에 넣어 조회하므로 기존 DB나 서버 설정이 필요 없습니다. 실습용 CREATE TABLE·INSERT와 조회용 SELECT를 구분하며, 실제 업무 DB에는 연결하지 않습니다. 운영 DB에서 읽기 권한·조회 범위를 확인하는 절차는 별도입니다.

실습 ZIP 받기: SQLite 매개변수 정확 일치 조회

1. 세 코드의 정답부터 적습니다

첨부의 fixtures/normal.csv는 다음 세 행입니다. 모두 합성 자료이며 실제 고객·상품 정보가 아닙니다.

code,label
A100,일반 코드
O'REILLY,따옴표 코드
' OR 1=1 --,SQL처럼 생긴 코드

조회하기 전에 어느 행이 나와야 하는지 정합니다. 코드를 SQL 구문으로 해석하지 않고 값 전체를 비교하는 것이 목표입니다.

마지막 값은 느낌표가 하나 더 붙어 등록된 값과 다릅니다. 0행은 조회에 성공했지만 일치하는 값이 없다는 정상 결과입니다. 반대로 따옴표나 SQL처럼 생긴 문자열을 조회했는데 세 행 전부가 나오면 이 실습의 완료 조건을 위반합니다. 행 수뿐 아니라 반환된 code와 label도 대조합니다.

2. SQL은 고정하고 값은 (query,)로 넘깁니다

핵심은 다음 두 줄입니다.

SELECT_SQL = 'SELECT code, label FROM items WHERE code = ? COLLATE BINARY'
rows = con.execute(SELECT_SQL, (query,)).fetchall()

?는 qmark 방식의 자리표시자입니다. 두 번째 인수 (query,)가 그 자리에 사용할 값 한 개를 전달합니다. ? 자체를 작은따옴표로 감싸지 않고, 조회 값을 SQL 문자열에 붙이거나 f-string·문자열 서식으로 넣지 않습니다. 값 속의 작은따옴표를 직접 이스케이프할 필요도 없습니다. Python sqlite3 자리표시자 문서

쉼표가 중요합니다. (query)는 한 항목 tuple이 아니라 괄호로 감싼 원래 문자열입니다. query='A100'을 그렇게 넘기면 문자는 네 개여서, 자리표시자 한 개에 값 네 개를 전달한 것으로 읽혀 sqlite3.ProgrammingError가 발생합니다. 문자열이 한 글자면 우연히 개수가 맞을 수 있으므로 “오류가 없었다”만으로 작성법이 맞다고 판단하지 않습니다. 이번 코드는 항상 (query,)를 사용합니다.

자리표시자는 값 위치를 위한 기능입니다. 테이블 이름·열 이름·정렬 방향을 넣는 기능으로 쓰지 않습니다. 이 코드에서는 items, code, label과 SQL 구조를 모두 고정했습니다. 식별자를 선택해야 하는 다른 프로그램이라면 입력 문자열을 곧바로 SQL에 넣지 말고, 허용 목록의 미리 정한 SQL을 고르는 설계를 별도로 해야 합니다. SQLite 매개변수 설명

비교는 =와 COLLATE BINARY를 사용합니다. a100과 A100, A100 과 A100을 구분하며 %와 _도 일반 값으로 찾습니다. 부분 검색용 LIKE는 사용하지 않습니다. 문자열 비교 규칙은 SQLite collating sequences 문서를 참고하세요.

3. 조회 값은 UTF-8 파일로 전달합니다

조회 값을 명령줄에 직접 입력하면 PowerShell·CMD·Bash의 따옴표 규칙까지 신경 써야 합니다. 이번에는 ZIP 안의 조회 파일을 그대로 전달합니다. 예를 들어 queries/apostrophe.txt에는 다음 값이 들어 있습니다.

O'REILLY

파일에 SQL용 따옴표를 추가하지 않습니다. queries/sql_literal.txt의 값도 다음 문자열 그대로입니다.

' OR 1=1 --

조회 파일의 처리 규칙을 먼저 확인하세요.

• BOM 없는 UTF-8이며 파일 한도는 4,096바이트, 조회 값 한도는 256자입니다
• 파일 끝의 LF 한 개 또는 CRLF 한 쌍만 제거합니다. 끝 줄바꿈이 없어도 같은 값으로 조회합니다
• 그 뒤 CR·LF가 남으면 여러 줄로 보고 거부합니다. 마지막 줄바꿈 두 개, 단독 CR도 거부합니다
• 빈 파일과 LF 한 개만 있는 파일은 모두 빈 문자열 조회입니다
• strip()으로 앞뒤 공백을 지우지 않고, 대소문자 변환이나 Unicode 정규화도 하지 않습니다
• BOM·NUL은 조회 값으로 받지 않습니다

따라서 A100 뒤의 줄바꿈은 저장 편의를 위한 구분자 하나로만 처리하고, A100 의 마지막 공백은 조회 값으로 유지합니다. 코드에 줄바꿈 자체를 포함해야 하는 업무라면 이 입력 계약을 그대로 사용할 수 없습니다.

CSV는 머리글이 순서까지 code,label이어야 하고 1~10,000 데이터 레코드, 1,000,000바이트까지 받습니다. code는 0~256자, label은 1~256자로 제한하며 둘 다 NUL·CR·LF를 거부합니다. 빈 code는 허용하되 중복 code는 거부합니다. BOM이 붙은 CSV 머리글은 BAD_HEADER로 중단합니다. 이는 이 실습의 입력 규칙입니다. Python CSV reader 문서

4. 압축을 풀고 정상 조회를 실행합니다

1. ZIP을 새 연습 폴더에 풀고 터미널에서 그 폴더로 이동합니다
2. parameter_query.py, fixtures, queries가 같은 폴더 아래에 있는지 확인합니다
3. 입력을 다른 프로그램에서 수정하지 않는 동안 실행합니다. 운영 자료 폴더는 사용하지 않습니다
4. 다음 명령으로 실행 환경을 확인합니다

python3 -c "import platform,sqlite3; print(platform.python_version()); print(sqlite3.sqlite_version)"

코드는 Python 3.12 이상을 요구합니다. 실제 검증한 조합은 Linux CPython 3.12.14 / SQLite 3.53.1입니다. Windows·macOS의 명령 실행과 앱 동작은 검증하지 않았습니다. python3가 없으면 설치된 Python의 실행 명령을 확인한 뒤 명령의 실행 파일 이름만 바꿉니다.

다음 결과 파일 이름은 아직 없어야 합니다. 출력 부모 폴더는 이미 있어야 하며, 여기서는 압축을 푼 현재 폴더에 결과를 만듭니다.

python3 parameter_query.py fixtures/normal.csv queries/a100.txt result-a100.json
python3 parameter_query.py fixtures/normal.csv queries/apostrophe.txt result-apostrophe.json
python3 parameter_query.py fixtures/normal.csv queries/sql_literal.txt result-sql-literal.json
python3 parameter_query.py fixtures/normal.csv queries/not_found.txt result-not-found.json

실제 실행에서 확인한 핵심 결과는 다음과 같습니다. 각 명령은 한 줄 JSON을 콘솔에 출력하고 같은 보고서를 새 JSON 파일에 저장합니다.

a100.txt        -> rows=[["A100", "일반 코드"]], matched_rows=1, 종료 코드 0
apostrophe.txt  -> rows=[["O'REILLY", "따옴표 코드"]], matched_rows=1, 종료 코드 0
sql_literal.txt -> rows=[["' OR 1=1 --", "SQL처럼 생긴 코드"]], matched_rows=1, 종료 코드 0
not_found.txt   -> rows=[], matched_rows=0, 종료 코드 0
공통: status=PASS, input_rows=3, bound_parameters=1

source_preserved와 query_preserved는 이 실행에서 읽은 원본 바이트와 출력 직전 재확인한 바이트의 SHA-256이 같았다는 뜻입니다. 보고서에는 두 SHA-256과 Python·SQLite 버전도 남습니다. 외부 프로그램의 모든 동시 교체나 공격을 막았다는 보안 판정은 아닙니다.

5. 전체 코드를 저장합니다

다음 전체 코드와 ZIP의 parameter_query.py는 같습니다. 일부 함수만 복사하지 말고 같은 파일 이름으로 저장하세요. 외부 패키지는 설치하지 않습니다.

"""Exact text lookup in a disposable SQLite practice database; Python 3.12+."""
import argparse
import csv
import hashlib
import io
import json
import os
import platform
import sqlite3
import stat
import sys
import tempfile
from contextlib import closing
from pathlib import Path

MAX_CSV_BYTES = 1_000_000
MAX_QUERY_BYTES = 4_096
MAX_ROWS = 10_000
MAX_TEXT_CHARS = 256
SCHEMA = '''CREATE TABLE items (
    code TEXT COLLATE BINARY PRIMARY KEY NOT NULL,
    label TEXT NOT NULL
)'''
INSERT_SQL = 'INSERT INTO items(code, label) VALUES(?, ?)'
SELECT_SQL = 'SELECT code, label FROM items WHERE code = ? COLLATE BINARY'


class PracticeError(Exception):
    def __init__(self, code, message):
        super().__init__(message)
        self.code = code


def digest(raw):
    return hashlib.sha256(raw).hexdigest()


def plain_path(value, io_code):
    path = Path(value)
    if '..' in path.parts:
        raise PracticeError('BAD_PATH', 'Do not use .. in practice paths')
    path = Path(os.path.abspath(path))
    for part in (*reversed(path.parents), path):
        try:
            info = part.lstat()
        except OSError as error:
            raise PracticeError(io_code, str(error)) from error
        if stat.S_ISLNK(info.st_mode):
            raise PracticeError('PATH_LINK', 'Symbolic links are not accepted')
        if getattr(info, 'st_reparse_tag', 0):
            raise PracticeError('PATH_LINK', 'Reparse points are not accepted')
    return path


def new_target(value):
    raw = Path(value)
    if raw.name in ('', '.', '..') or '..' in raw.parts:
        raise PracticeError('BAD_PATH', 'Specify a new output file without ..')
    parent = plain_path(raw.parent, 'OUTPUT_IO')
    if not parent.is_dir():
        raise PracticeError('OUTPUT_IO', 'Output parent must already be a directory')
    target = parent / raw.name
    if os.path.lexists(target):
        raise PracticeError('OUTPUT_EXISTS', 'Choose a new output file name')
    return target


def read_input(value, limit):
    path = plain_path(value, 'INPUT_IO')
    info = path.stat()
    if not stat.S_ISREG(info.st_mode):
        raise PracticeError('INPUT_IO', 'Input must be a regular file')
    if info.st_nlink != 1:
        raise PracticeError('INPUT_LINK', 'Hard-linked input files are not accepted')
    try:
        with path.open('rb') as handle:
            raw = handle.read(limit + 1)
    except OSError as error:
        raise PracticeError('INPUT_IO', str(error)) from error
    if len(raw) > limit:
        raise PracticeError('INPUT_LIMIT', 'Input exceeds its byte limit')
    return path, raw


def utf8(raw):
    try:
        return raw.decode('utf-8', errors='strict')
    except UnicodeError as error:
        raise PracticeError('BAD_ENCODING', 'Save a UTF-8 file without BOM') from error


def load_records(raw):
    try:
        rows = list(csv.reader(io.StringIO(utf8(raw), newline=''), strict=True))
    except csv.Error as error:
        raise PracticeError('BAD_CSV', str(error)) from error
    if not rows or rows[0] != ['code', 'label']:
        raise PracticeError('BAD_HEADER', 'Header must be exactly code,label')
    records = rows[1:]
    if not records:
        raise PracticeError('EMPTY_CSV', 'At least one data record is required')
    if len(records) > MAX_ROWS:
        raise PracticeError('INPUT_LIMIT', 'CSV exceeds 10,000 data records')
    seen = set()
    for number, fields in enumerate(records, start=2):
        if len(fields) != 2:
            raise PracticeError('BAD_ROW', 'Record ' + str(number) + ': two fields required')
        code, label = fields
        if (len(code) > MAX_TEXT_CHARS or not 1 <= len(label) <= MAX_TEXT_CHARS
                or any(char in code + label for char in '\x00\r\n')):
            raise PracticeError('BAD_VALUE', 'Record ' + str(number) + ': invalid text')
        if code in seen:
            raise PracticeError('DUPLICATE_CODE', 'Record ' + str(number) + ': duplicate code')
        seen.add(code)
    return records


def load_query(raw):
    query = utf8(raw)
    if query.endswith('\r\n'):
        query = query[:-2]
    elif query.endswith('\n'):
        query = query[:-1]
    if (len(query) > MAX_TEXT_CHARS or query.startswith('\ufeff')
            or any(char in query for char in '\x00\r\n')):
        raise PracticeError('BAD_QUERY', 'Use one text value, at most one final LF or CRLF')
    return query


def confirm_unchanged(path, expected, limit):
    _, raw = read_input(path, limit)
    if digest(raw) != expected:
        raise PracticeError('INPUT_CHANGED', 'Input changed; retry from a stable copy')


def publish_new(target, report):
    stage = None
    try:
        fd, name = tempfile.mkstemp(prefix='.parameter-query-', suffix='.tmp', dir=target.parent)
        stage = Path(name)
        with os.fdopen(fd, 'wb') as handle:
            handle.write((json.dumps(report, ensure_ascii=False, indent=2) + '\n').encode('utf-8'))
            handle.flush()
            os.fsync(handle.fileno())
        if json.loads(stage.read_text(encoding='utf-8')) != report:
            raise PracticeError('OUTPUT_IO', 'Temporary JSON readback differs')
        # Link the complete staged file to a new name; never replace an existing name.
        os.link(stage, target)
    except FileExistsError as error:
        raise PracticeError('OUTPUT_EXISTS', 'Output appeared; choose a new name') from error
    except OSError as error:
        raise PracticeError('OUTPUT_IO', str(error)) from error
    finally:
        if stage is not None:
            try:
                stage.unlink()
            except OSError as error:
                print('TEMP_CLEANUP_WARNING: ' + str(error), file=sys.stderr)


def run(csv_file, query_file, output_file):
    if sys.version_info < (3, 12):
        raise PracticeError('PYTHON_VERSION', 'Python 3.12 or later is required')
    target = new_target(output_file)
    source, csv_raw = read_input(csv_file, MAX_CSV_BYTES)
    query_path, query_raw = read_input(query_file, MAX_QUERY_BYTES)
    records = load_records(csv_raw)
    query = load_query(query_raw)
    with closing(sqlite3.connect(':memory:')) as con:
        with con:
            con.execute(SCHEMA)
            con.executemany(INSERT_SQL, records)
            rows = con.execute(SELECT_SQL, (query,)).fetchall()
    if len(rows) > 1:
        raise PracticeError('RESULT_SHAPE', 'Expected zero or one exact matching row')
    source_hash, query_hash = digest(csv_raw), digest(query_raw)
    confirm_unchanged(source, source_hash, MAX_CSV_BYTES)
    confirm_unchanged(query_path, query_hash, MAX_QUERY_BYTES)
    report = {'status': 'PASS', 'query': query, 'input_rows': len(records),
              'matched_rows': len(rows), 'rows': [list(row) for row in rows],
              'bound_parameters': 1, 'source_sha256': source_hash,
              'query_sha256': query_hash, 'source_preserved': True,
              'query_preserved': True, 'python': platform.python_version(),
              'sqlite': sqlite3.sqlite_version, 'real_database_access': False}
    publish_new(target, report)
    return report


def main():
    parser = argparse.ArgumentParser(description='Look up one UTF-8 text value in a practice memory DB')
    parser.add_argument('csv_file')
    parser.add_argument('query_file')
    parser.add_argument('new_output_file')
    args = parser.parse_args()
    try:
        report = run(args.csv_file, args.query_file, args.new_output_file)
    except (PracticeError, OSError, sqlite3.Error) as error:
        print(json.dumps({'status': 'REJECTED', 'error_code': getattr(error, 'code', 'RUNTIME_ERROR'),
                          'message': str(error)}, ensure_ascii=False))
        return 2
    print(json.dumps(report, ensure_ascii=False))
    return 0


if __name__ == '__main__':
    sys.exit(main())

sqlite3.connect(':memory:')는 메모리 DB 연결입니다. 이 코드가 수행하는 생성·삽입은 가상 CSV를 조회할 연습 표를 준비하기 위한 것입니다. 연결은 closing(...)으로 닫으며 DB 파일을 남기지 않습니다. 메모리 DB를 만드는 이 부분을 운영 DB 연결 코드로 단순 교체해 사용하지 마세요. Python sqlite3 연결 문서

출력은 같은 폴더의 임시 파일에 완성한 뒤 flush·fsync와 JSON 재읽기 대조를 거쳐 새 이름으로 연결합니다. 기존 파일을 replace하지 않습니다. 이 방식에는 하드 링크를 지원하는 로컬 파일시스템이 필요하며, 연결이 실패하면 OUTPUT_IO로 중단합니다. Python os.link 문서

6. 실패와 “찾지 못함”을 나누어 확인합니다

다음 명령은 의도적으로 잘못된 입력을 사용합니다. 각각 새 결과 이름으로 실행합니다.

python3 parameter_query.py fixtures/duplicate.csv queries/a100.txt result-duplicate.json
python3 parameter_query.py fixtures/bad_header.csv queries/a100.txt result-header.json
python3 parameter_query.py fixtures/normal.csv queries/bad_utf8.txt result-encoding.json
python3 parameter_query.py fixtures/normal.csv queries/double_eol.txt result-lines.json

위 네 실패 입력은 실제 시험에서 각각 지정한 오류 코드, 종료 코드 2로 거부됐고 새 최종 결과 파일이 남지 않았습니다. 정상 조회의 0행·종료 코드 0과 구별하세요. 같은 정상 명령을 두 번 실행하면 두 번째는 OUTPUT_EXISTS로 거부하며 첫 결과의 해시가 그대로인지 테스트가 확인합니다.

파일·부모 폴더가 없거나 저장소에서 하드 링크를 만들 수 없으면 INPUT_IO 또는 OUTPUT_IO가 나옵니다. 심볼릭 링크·검출 가능한 reparse point·하드 링크된 입력은 사용하지 않으며, ..가 들어 있는 경로도 거부합니다. 링크 검사를 끄기보다 일반 로컬 파일과 정상 경로의 새 연습 폴더에서 다시 실행합니다.

인수 개수가 틀렸을 때는 argparse가 stderr에 사용법을 표시하고 종료 코드 2로 끝납니다. 이 경우 오류 JSON은 출력하지 않습니다. ? 한 개에 두 값을 전달하거나 (query,)의 쉼표를 빠뜨리는 실수는 첨부 테스트에서 별도로 ProgrammingError를 확인합니다.

7. 경계값과 첨부 테스트까지 실행합니다

python3 test_parameter_query.py

최종 로컬 실행은 58개 테스트, 실패 0·오류 0·건너뜀 0이었습니다. 기본 세 값 외에 다음 차이도 확인했습니다.

• normal.csv에서 빈 값·a100· A100 ·%·_·작은따옴표 하나가 더 있는 변형은 모두 0행입니다
• boundary.csv에서는 빈 code, 한글 code, 앞뒤 공백이 있는 code, 소문자 code가 각각 자기 행을 반환합니다
• 조합형 é와 e + U+0301은 서로 다른 두 code로 보존되어 각각 대응하는 label 한 개를 반환합니다
• 끝 줄바꿈 없음·LF·CRLF는 같은 A100을 조회하며, 여러 줄·두 번의 끝 줄바꿈·단독 CR은 거부합니다
• 중복·열 수·CSV 구문·UTF-8·BOM·제어문자·길이·입력 크기·레코드 한도와 기존 출력·링크 경로도 검사합니다

expected/cases.json의 정답은 코드를 쓰기 전에 가상 입력에서 직접 명시했습니다. 결과의 rows·input_rows는 이 기대값과 비교하고, matched_rows는 기대 목록 길이와 비교합니다. 입력 SHA-256은 현재 파일 바이트의 해시와 대조하며 버전 필드는 실제 테스트 실행기의 버전과 대조합니다. 콘솔 JSON과 파일 JSON은 모든 필드를 포함해 비교합니다. 환경마다 달라질 수 있는 버전이나 해시를 행 결과의 고정 정답으로 대신 쓰지 않습니다.

테스트는 임시 폴더에서만 결과를 만들고 fixtures·queries 전체의 SHA-256이 실행 전후 같은지도 확인합니다. 입력 변경·출력 이름 선점·하드 링크 실패는 통제된 코드 단계 시험입니다. 실제 장애가 일어났다고 꾸민 기록이 아닙니다.

적용 범위와 한계

이 실습은 고정된 한 SELECT의 값 바인딩과 정확 일치 조회에 한정합니다. CSV를 영구 SQLite 파일에 적재하는 절차와 트랜잭션 제약은 CSV를 로컬 SQLite에 넣고 중복 ID와 수량 오류 거부하기에서 이어서 볼 수 있습니다. JOIN 뒤 누락·중복·NULL을 대조하려면 AI가 만든 SQL에서 행 수와 NULL 처리 확인하기의 검산 목표가 필요합니다.

조회 값 바인딩만으로 운영 권한, 개인정보 처리, 모든 SQL 구조, 다른 DB 드라이버의 자리표시자 방식까지 해결되지는 않습니다. 실제 DB에는 조회 권한과 범위를 따로 정하고, 이 코드의 실습 표 생성 부분을 넘기지 않습니다. 대용량 성능·실행 계획·Windows 파일 잠금·네트워크 저장소 동작은 측정하지 않았습니다.

입력과 경로를 동시에 바꾸는 다른 프로그램이 없는 격리된 개인 연습 폴더를 전제로 합니다. 파일 경로 확인은 적대적인 동시 변경을 막는 보안 도구가 아닙니다. 전원 손실이나 프로세스 강제 종료를 시험하지 않았으며, 그런 경우 임시 파일이 남을 수 있습니다. 일반 정리 실패 시에는 TEMP_CLEANUP_WARNING이 stderr에 표시됩니다. 완성된 결과 파일이 이미 만들어졌다면 이 경고를 불완전한 결과와 혼동하지 말고 임시 파일 상태부터 확인하세요.

자주 묻는 질문

따옴표를 두 번 붙여 저장해야 하나요?

아니요. CSV와 조회 파일에는 O'REILLY를 그대로 둡니다. SQL 문장에 값을 붙이지 않고 (query,)로 전달합니다.

빈 문자열은 NULL인가요?

다릅니다. 이번 fixture는 문자열 code만 다룹니다. 빈 문자열을 허용한 boundary.csv에서는 빈 code 한 행을 찾고, normal.csv에서는 0행입니다. NULL 조회 정책은 이번 범위에 없습니다.

테이블 이름도 ?에 넣을 수 있나요?

이 자리표시자는 값 용도입니다. 이 실습의 테이블·열 이름은 고정되어 있습니다. 식별자가 가변인 프로그램은 허용 목록과 SQL 구조를 별도로 설계해야 합니다.

결과가 0행이면 실패인가요?

등록된 값과 정확히 같지 않으면 0행도 PASS, 종료 코드 0입니다. 파일·인코딩·중복·인수 오류로 거부한 종료 코드 2와 구분합니다.

공식 문서 확인일은 2026-10-07 KST입니다. Python 3.12 계열 문서를 확인했으며 문서 페이지의 최신 패치 버전과 실제 실행 버전은 따로 기록했습니다. 실습 파일에는 원본·조회 파일·독립 기대값·전체 코드·테스트·README·SHA-256 체크섬이 들어 있습니다.

작성자 소개

최지원

엑셀·VBA·파이썬·AI 업무 자동화의 사용 절차와 실습 자료를 정리합니다.

WP 게시물 작성자의 글 보기