Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion pandas/io/sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,30 @@
# -- Helper functions


def _sa_text_if_string(stmt: Any) -> Any:
"""
Wrap plain SQL strings with sqlalchemy.text() if SQLAlchemy is available.

Parameters
----------
stmt : Any
The SQL statement or object.

Returns
-------
Any
`sqlalchemy.sql.elements.TextClause` if wrapping occurred,
otherwise the original statement.
"""
try:
import sqlalchemy as sa # lazy import; keep SA optional
except ImportError:
return stmt
return sa.text(stmt) if isinstance(stmt, str) else stmt


def _process_parse_dates_argument(parse_dates):

"""Process parse_dates argument for read_sql functions"""
# handle non-list entries for parse_dates gracefully
if parse_dates is True or parse_dates is None or parse_dates is False:
Expand Down Expand Up @@ -1853,7 +1876,8 @@ def read_query(
read_sql

"""
result = self.execute(sql, params)
stmt = _sa_text_if_string(sql)
result = self.execute(stmt, params)
columns = result.keys()

if chunksize is not None:
Expand Down Expand Up @@ -1882,6 +1906,7 @@ def read_query(
)
return frame


read_sql = read_query

def prep_table(
Expand Down
27 changes: 27 additions & 0 deletions pandas/tests/io/sql/test_percent_patterns.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
from __future__ import annotations

from typing import Any

import pytest

pytest.importorskip("sqlalchemy")


def test_modulo_operator(sql_con: Any) -> None:
query = "SELECT 10 % 3"
result = sql_con.execute(query)
assert result.scalar() == 1


def test_like_pattern(sql_con: Any) -> None:
query = "SELECT 'abc' LIKE 'a%'"
result = sql_con.execute(query)
assert result.scalar() == 1


def test_sqlalchemy_selectable(sql_con: Any) -> None:
from sqlalchemy import literal, select

stmt = select(literal("hello"))
result = sql_con.execute(stmt)
assert result.scalar() == "hello"
Loading