浅念
8cea7dee20
- SiteForge Enterprise V8.10.0-dev.1 upstream source (210 tests) - Fix: v8_workflows._mapped_input now reads both 类型/分类 columns so WXR imports keep post/product types instead of collapsing to page - Docs: trade-pro theme port spec (R3) and implementation plan Co-Authored-By: Claude Code <noreply@anthropic.com>
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
"""PostgreSQL DB-API bridge for legacy '?' SQL (optional, requires psycopg[binary]).
|
|
|
|
No automatic conversion of existing SQLite files is attempted. Production migration
|
|
requires a tested one-time data migration and reconciliation.
|
|
"""
|
|
from __future__ import annotations
|
|
import re
|
|
from contextlib import contextmanager
|
|
|
|
|
|
def placeholders(sql: str) -> str:
|
|
"""Translate SQLite positional placeholders outside SQL quoted strings."""
|
|
out=[];quote=None;i=0
|
|
while i<len(sql):
|
|
ch=sql[i]
|
|
if quote:
|
|
out.append(ch)
|
|
if ch==quote:
|
|
if i+1<len(sql) and sql[i+1]==quote:
|
|
out.append(sql[i+1]);i+=1
|
|
else:quote=None
|
|
elif ch in ("'",'"'):
|
|
quote=ch;out.append(ch)
|
|
elif ch=='?':out.append('%s')
|
|
else:out.append(ch)
|
|
i+=1
|
|
return ''.join(out)
|
|
|
|
|
|
def normalize(sql: str) -> str:
|
|
sql=sql.strip()
|
|
if re.match(r'^INSERT\s+OR\s+IGNORE\s+INTO\b',sql,re.I):
|
|
sql=re.sub(r'^INSERT\s+OR\s+IGNORE\s+INTO\b','INSERT INTO',sql,flags=re.I)
|
|
sql=sql.rstrip(';')+' ON CONFLICT DO NOTHING'
|
|
return placeholders(sql)
|
|
|
|
|
|
class PGConnection:
|
|
def __init__(self,dsn):
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
self.connection=psycopg.connect(dsn,row_factory=dict_row,connect_timeout=5)
|
|
self.total_changes=0
|
|
|
|
def execute(self,sql,params=()):
|
|
stripped=sql.strip()
|
|
if stripped.upper().startswith('PRAGMA TABLE_INFO('):
|
|
table=stripped.split('(',1)[1].split(')',1)[0].strip().strip('"\'')
|
|
if not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*',table):raise ValueError('Invalid table name')
|
|
return self.connection.execute('SELECT column_name AS name FROM information_schema.columns WHERE table_schema=current_schema() AND table_name=%s',(table,))
|
|
if stripped.upper()=='BEGIN IMMEDIATE':
|
|
# Conservative transaction-wide lock while porting SQLite queue semantics.
|
|
# Replace with per-site advisory locks before large multi-worker scale.
|
|
return self.connection.execute('SELECT pg_advisory_xact_lock(823947231)')
|
|
if stripped.upper().startswith('PRAGMA '):
|
|
return _EmptyResult()
|
|
cursor=self.connection.execute(normalize(sql),params)
|
|
if cursor.rowcount and cursor.rowcount>0 and stripped.split(None,1)[0].upper() in ('UPDATE','INSERT','DELETE'):
|
|
self.total_changes+=cursor.rowcount
|
|
return cursor
|
|
|
|
def executescript(self,script):
|
|
# Schema migrations contain only simple DDL and SQLite PRAGMA statements.
|
|
for statement in script.split(';'):
|
|
if statement.strip():self.execute(statement)
|
|
|
|
def __enter__(self):return self
|
|
def __exit__(self,typ,val,tb):
|
|
try:
|
|
if typ:self.connection.rollback()
|
|
else:self.connection.commit()
|
|
finally:self.connection.close()
|
|
return False
|
|
|
|
class _EmptyResult:
|
|
rowcount=0
|
|
def fetchone(self):return None
|
|
def fetchall(self):return []
|
|
|
|
|
|
def connect(dsn):return PGConnection(dsn)
|