fix(persistence): resolve resource URL migration and scoped SQL failures

- Reflect migration tables without resolving foreign-key targets
- Allow validated resource URL expansion through the tenant SQL AST guard
- Add regression coverage for trusted and malformed expansion expressions
This commit is contained in:
BiFangKNT
2026-10-23 14:39:38 +08:00
parent 97204a5d94
commit 5d774ff555
3 changed files with 32 additions and 10 deletions
@@ -35,7 +35,8 @@ def upgrade():
if not retained:
continue
# Reflection intentionally reads literal stored values without expansion.
source = sa.Table(table.name, sa.MetaData(), autoload_with=connection)
# Only local columns are needed; historical schemas may omit FK targets.
source = sa.Table(table.name, sa.MetaData(), autoload_with=connection, resolve_fks=False)
retained = [name for name in retained if name in source.c]
keys = list(source.primary_key.columns)
if len(keys) != 1:
+9 -9
View File
@@ -412,8 +412,16 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s
'TenantUnitOfWork public SQL must use a structured SQLAlchemy query, values, or DML statement'
)
from .resource_urls import ExpandedResourceURLs
for element in _iter_scoped_statement_elements(statement):
if not type(element).__module__.startswith('sqlalchemy.'):
resource_expansion_allowed = (
type(element) is ExpandedResourceURLs
and len(list(element.clauses)) == 1
and isinstance(list(element.clauses)[0], sqlalchemy.Column)
and type(list(element.clauses)[0].type) is MediaReferenceText
)
if not type(element).__module__.startswith('sqlalchemy.') and not resource_expansion_allowed:
raise ScopedSessionTransactionError(
'TenantUnitOfWork does not allow custom SQL AST nodes in public statements'
)
@@ -503,14 +511,6 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s
)
if isinstance(element, sqlalchemy.sql.functions.FunctionElement):
from .resource_urls import ExpandedResourceURLs
resource_expansion_allowed = (
type(element) is ExpandedResourceURLs
and len(list(element.clauses)) == 1
and isinstance(list(element.clauses)[0], sqlalchemy.Column)
and type(list(element.clauses)[0].type) is MediaReferenceText
)
function_name = str(getattr(element, 'name', '')).casefold()
package_names = tuple(getattr(element, 'packagenames', ()))
generic_function_allowed = (