from __future__ import annotations
from textwrap import dedent
from typing import TYPE_CHECKING
from .input import strparsing
from .models import dialect as _dialect
from .models import overrides as _overrides
from .models.dialect import DefaultDialect, SQLDialect
from .models.query import Query
if TYPE_CHECKING:
import csql
import csql.dialect
import csql.overrides
[docs]
def Q(
sql: str,
dialect: csql.dialect.SQLDialect
| csql.dialect.InferOrDefault = _dialect.InferOrDefault(DefaultDialect), # noqa: B008
overrides: csql.overrides.Overrides
| None
| csql.overrides.InferOrDefault = _overrides.InferOrDefault(None), # noqa: B008
) -> csql.Query:
"""
Create a :class:`csql.Query`.
Usage:
>>> p = Parameters(created_on=date(2020,1,1))
>>> q_cust = Q(f'''select name, customer_type from customers where created_on > {p['created_on']}''')
>>> q_count = Q(f'select customer_type, count(*) from {q_cust} group by rollup(type)')
See: :ref:`basic_usage`
:param sql: A string with a SQL query. The string is designed to be built with an ``f'f-string'``,
so you can interpolate Parameters and other Queries inside in a natural way.
:param dialect: A default :class:`dialect<csql.dialect.SQLDialect>` to use when building this Query.
By default, if this Query references another Query, the references Query's dialects will be used.
:param overrides: A default set of :class:`overrides<csql.overrides.Overrides>` to use when building this Query.
By default, if this Query references another Query, the references Query's overrides will be used.
"""
if callable(sql):
raise TypeError(
dedent("""
Passing a lambda to Q is no longer supported! You can now just pass an interpolated string directly:
Q(f"select from {blah}")
""").strip()
)
queryParts = strparsing.getQueryParts(sql)
existing_dialects = {
q.default_dialect
for q in queryParts
if isinstance(q, Query) and isinstance(q.default_dialect, SQLDialect)
}
existing_overrides = {
q.default_overrides for q in queryParts if isinstance(q, Query)
}
if isinstance(dialect, _dialect.InferOrDefault):
if len(existing_dialects) == 0:
dialect = dialect # noqa: PLW0127
elif len(existing_dialects) == 1:
dialect = next(iter(existing_dialects))
else:
ds = ", ".join(str(d) for d in existing_dialects)
raise Exception( # noqa: TRY002
dedent(f"""
Found multiple dialects when inferring the default:
{ds}.
If this is intentional, please specify a dialect for this query explicitly, e.g.
Q('select ...', dialect=csql.dialect.DefaultDialect)
""").strip()
)
if isinstance(overrides, _overrides.InferOrDefault):
if len(existing_overrides) == 0:
overrides = overrides # noqa: PLW0127
elif len(existing_overrides) == 1:
overrides = next(iter(existing_overrides))
else:
os = ", ".join(str(o) for o in existing_overrides)
raise Exception( # noqa: TRY002
dedent(f"""
Found multiple overrides when inferring the default:
{os}.
If this is intentional, please specify overrides for this query explicitly, e.g.
Q('select ...', overrides=None)
""").strip()
)
return Query(
queryParts=queryParts,
default_dialect=dialect,
default_overrides=overrides,
_extensions=frozenset(),
)