Compare commits

..

27 Commits

Author SHA1 Message Date
Sebastián Ramírez
4590963e88 🔖 Release version 0.0.19 2024-06-03 22:26:53 -05:00
github-actions
883cbe3a8d 📝 Update release notes 2024-06-04 03:22:28 +00:00
Sebastián Ramírez
b560e9deb8 ⬆️ Upgrade Ruff and Black (#968)
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2024-06-03 22:22:04 -05:00
github-actions
e2f646dea5 📝 Update release notes 2024-06-04 03:19:04 +00:00
dependabot[bot]
b93dd95125 ⬆ Bump tiangolo/issue-manager from 0.4.1 to 0.5.0 (#922)
Bumps [tiangolo/issue-manager](https://github.com/tiangolo/issue-manager) from 0.4.1 to 0.5.0.
- [Release notes](https://github.com/tiangolo/issue-manager/releases)
- [Commits](https://github.com/tiangolo/issue-manager/compare/0.4.1...0.5.0)

---
updated-dependencies:
- dependency-name: tiangolo/issue-manager
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-06-03 22:18:48 -05:00
github-actions
ceac7bc2e8 📝 Update release notes 2024-06-04 02:48:00 +00:00
Esteban Maya
1d43bd8b1e 🐛 Fix pydantic EmailStr support and max_length in several String subclasses (#966) 2024-06-03 21:47:40 -05:00
github-actions
9f3af8507e 📝 Update release notes 2024-06-04 02:35:16 +00:00
Sebastián Ramírez
d165e4b5ad ♻️ Refactor generate select template to isolate templated code to the minimum (#967)
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2024-06-03 21:34:54 -05:00
github-actions
d5cba6e358 📝 Update release notes 2024-06-04 01:39:25 +00:00
Sebastián Ramírez
bd1641c9a2 ⬆️ Update minimum SQLAlchemy version to 2.0.14 as that one includes TryCast used internally (#964) 2024-06-03 20:39:07 -05:00
github-actions
71de44daba 📝 Update release notes 2024-06-04 01:34:41 +00:00
Sebastián Ramírez
1b275bd6a7 📌 Pin typing-extensions in tests for compatiblity with Python 3.8, dirty-equals, Pydantic (#965) 2024-06-04 01:34:21 +00:00
github-actions
866d9ecb29 📝 Update release notes 2024-06-04 00:39:40 +00:00
Esteban Maya
5bb4cffd49 🐛 Fix set varchar limit when max_length is set on Pydantic models using Pydantic v2 (#963)
Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
2024-06-04 00:39:23 +00:00
github-actions
a319952be1 📝 Update release notes 2024-06-03 23:56:49 +00:00
Soof Golan
662bd641b8 ✏️ Fix broken link to @dataclass_transform (now PEP 681) in docs/features.md (#753)
Co-authored-by: Sebastián Ramírez <tiangolo@gmail.com>
Co-authored-by: Patrick Arminio <patrick.arminio@gmail.com>
2024-06-03 23:56:30 +00:00
github-actions
dcf4f58e81 📝 Update release notes 2024-05-10 21:00:41 +00:00
Sebastián Ramírez
e4013acc54 👷 Update GitHub Actions to download and upload artifacts (#936) 2024-05-10 21:00:24 +00:00
github-actions
df0f834227 📝 Update release notes 2024-05-07 18:32:44 +00:00
Sebastián Ramírez
5e592c9a0d 👷 Tweak CI for test-redistribute, add needed env vars for slim (#929) 2024-05-07 18:32:16 +00:00
Sebastián Ramírez
c13b71056e 📝 Update release notes 2024-04-29 23:29:21 -07:00
Sebastián Ramírez
900e0d3371 🔖 Release version 0.0.18 2024-04-29 23:25:02 -07:00
github-actions
a280b58c10 📝 Update release notes 2024-04-30 06:22:46 +00:00
Sebastián Ramírez
9ebbf255f7 Add sqlmodel-slim setup (#916) 2024-04-30 06:22:28 +00:00
github-actions
39cbf27904 📝 Update release notes 2024-04-30 00:01:02 +00:00
Sebastián Ramírez
28d0e76370 🔧 Re-enable MkDocs Material Social plugin (#915) 2024-04-30 00:00:40 +00:00
29 changed files with 682 additions and 802 deletions

View File

@@ -71,7 +71,7 @@ jobs:
run: python ./scripts/docs.py verify-readme
- name: Build Docs
run: python ./scripts/docs.py build
- uses: actions/upload-artifact@v3
- uses: actions/upload-artifact@v4
with:
name: docs-site
path: ./site/**

View File

@@ -19,18 +19,16 @@ jobs:
run: |
rm -rf ./site
mkdir ./site
- name: Download Artifact Docs
id: download
uses: dawidd6/action-download-artifact@v2.28.0
- uses: actions/download-artifact@v4
with:
if_no_artifact_found: ignore
github_token: ${{ secrets.GITHUB_TOKEN }}
workflow: build-docs.yml
run_id: ${{ github.event.workflow_run.id }}
name: docs-site
path: ./site/
pattern: docs-site
merge-multiple: true
github-token: ${{ secrets.GITHUB_TOKEN }}
run-id: ${{ github.event.workflow_run.id }}
- name: Deploy to Cloudflare Pages
if: steps.download.outputs.found_artifact == 'true'
# hashFiles returns an empty string if there are no files
if: hashFiles('./site/*')
id: deploy
uses: cloudflare/pages-action@v1
with:

View File

@@ -18,7 +18,7 @@ jobs:
issue-manager:
runs-on: ubuntu-latest
steps:
- uses: tiangolo/issue-manager@0.4.1
- uses: tiangolo/issue-manager@0.5.0
with:
token: ${{ secrets.GITHUB_TOKEN }}
config: >

View File

@@ -18,6 +18,7 @@ jobs:
matrix:
package:
- sqlmodel
- sqlmodel-slim
permissions:
id-token: write
steps:

View File

@@ -20,12 +20,14 @@ jobs:
- run: pip install smokeshow
- uses: dawidd6/action-download-artifact@v2.28.0
- uses: actions/download-artifact@v4
with:
workflow: test.yml
commit: ${{ github.event.workflow_run.head_sha }}
name: coverage-html
path: htmlcov
github-token: ${{ secrets.GITHUB_TOKEN }}
run-id: ${{ github.event.workflow_run.id }}
- run: smokeshow upload coverage-html
- run: smokeshow upload htmlcov
env:
SMOKESHOW_GITHUB_STATUS_DESCRIPTION: Coverage {coverage-percentage}
SMOKESHOW_GITHUB_COVERAGE_THRESHOLD: 95

View File

@@ -16,6 +16,7 @@ jobs:
matrix:
package:
- sqlmodel
- sqlmodel-slim
steps:
- name: Dump GitHub context
env:
@@ -40,6 +41,8 @@ jobs:
run: |
cd dist/sqlmodel*/
pip install -r requirements-tests.txt
env:
TIANGOLO_BUILD_PACKAGE: ${{ matrix.package }}
- name: Run source distribution tests
run: |
cd dist/sqlmodel*/

View File

@@ -51,7 +51,7 @@ jobs:
id: cache
with:
path: ${{ env.pythonLocation }}
key: ${{ runner.os }}-python-${{ env.pythonLocation }}-${{ hashFiles('pyproject.toml', 'requirements-tests.txt') }}
key: ${{ runner.os }}-python-${{ env.pythonLocation }}-${{ hashFiles('pyproject.toml', 'requirements-tests.txt') }}-v01
- name: Install Dependencies
if: steps.cache.outputs.cache-hit != 'true'
run: pip install -r requirements-tests.txt
@@ -60,7 +60,7 @@ jobs:
run: pip install --upgrade "pydantic>=1.10.0,<2.0.0"
- name: Install Pydantic v2
if: matrix.pydantic-version == 'pydantic-v2'
run: pip install --upgrade "pydantic>=2.0.2,<3.0.0"
run: pip install --upgrade "pydantic>=2.0.2,<3.0.0" "typing-extensions==4.6.1"
- name: Lint
# Do not run on Python 3.7 as mypy behaves differently
if: matrix.python-version != '3.7' && matrix.pydantic-version == 'pydantic-v2'
@@ -72,9 +72,9 @@ jobs:
COVERAGE_FILE: coverage/.coverage.${{ runner.os }}-py${{ matrix.python-version }}
CONTEXT: ${{ runner.os }}-py${{ matrix.python-version }}
- name: Store coverage files
uses: actions/upload-artifact@v3
uses: actions/upload-artifact@v4
with:
name: coverage
name: coverage-${{ matrix.python-version }}-${{ matrix.pydantic-version }}
path: coverage
coverage-combine:
needs:
@@ -89,10 +89,11 @@ jobs:
python-version: '3.8'
- name: Get coverage files
uses: actions/download-artifact@v3
uses: actions/download-artifact@v4
with:
name: coverage
pattern: coverage-*
path: coverage
merge-multiple: true
- run: pip install coverage[toml]
@@ -102,7 +103,7 @@ jobs:
- run: coverage html --show-contexts --title "Coverage for ${{ github.sha }}"
- name: Store coverage HTML
uses: actions/upload-artifact@v3
uses: actions/upload-artifact@v4
with:
name: coverage-html
path: htmlcov

View File

@@ -14,7 +14,7 @@ repos:
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.2.0
rev: v0.4.7
hooks:
- id: ruff
args:

View File

@@ -36,20 +36,10 @@ You will get completion for everything while writing the **minimum** amount of c
You won't need to keep guessing the types of different attributes in your models, if they could be `None`, etc. Your editor will be able to help you with everything because **SQLModel** is based on **standard Python type annotations**.
**SQLModel** even adopts currently <a href="https://github.com/microsoft/pyright/blob/main/specs/dataclass_transforms.md" class="external-link" target="_blank">in development standards</a> for Python type annotations to ensure the **best developer experience**, so you will get inline errors and autocompletion even while creating new model instances.
**SQLModel** adopts <a href="https://peps.python.org/pep-0681/" class="external-link" target="_blank">PEP 681</a> for Python type annotations to ensure the **best developer experience**, so you will get inline errors and autocompletion even while creating new model instances.
<img class="shadow" src="/img/index/autocompletion01.png">
/// info
Don't worry, adopting this in-development standard only affects/improves editor support.
It doesn't affect performance or correctness. And if the in-progress standard was deprecated your code won't be affected.
Meanwhile, you will get inline errors (like type checks) and autocompletion on places you wouldn't get with any other library. 🎉
///
## Short
**SQLModel** has **sensible defaults** for everything, with **optional configurations** everywhere.

View File

@@ -2,6 +2,43 @@
## Latest Changes
## 0.0.19
### Fixes
* 🐛 Fix pydantic `EmailStr` support and `max_length` in several String subclasses. PR [#966](https://github.com/tiangolo/sqlmodel/pull/966) by [@estebanx64](https://github.com/estebanx64).
* 🐛 Fix set varchar limit when `max_length` is set on Pydantic models using Pydantic v2. PR [#963](https://github.com/tiangolo/sqlmodel/pull/963) by [@estebanx64](https://github.com/estebanx64).
### Refactors
* ♻️ Refactor generate select template to isolate templated code to the minimum. PR [#967](https://github.com/tiangolo/sqlmodel/pull/967) by [@tiangolo](https://github.com/tiangolo).
### Upgrades
* ⬆️ Update minimum SQLAlchemy version to 2.0.14 as that one includes `TryCast` used internally. PR [#964](https://github.com/tiangolo/sqlmodel/pull/964) by [@tiangolo](https://github.com/tiangolo).
### Docs
* ✏️ Fix broken link to `@dataclass_transform` (now PEP 681) in `docs/features.md`. PR [#753](https://github.com/tiangolo/sqlmodel/pull/753) by [@soof-golan](https://github.com/soof-golan).
### Internal
* ⬆️ Upgrade Ruff and Black. PR [#968](https://github.com/tiangolo/sqlmodel/pull/968) by [@tiangolo](https://github.com/tiangolo).
* ⬆ Bump tiangolo/issue-manager from 0.4.1 to 0.5.0. PR [#922](https://github.com/tiangolo/sqlmodel/pull/922) by [@dependabot[bot]](https://github.com/apps/dependabot).
* 📌 Pin typing-extensions in tests for compatiblity with Python 3.8, dirty-equals, Pydantic. PR [#965](https://github.com/tiangolo/sqlmodel/pull/965) by [@tiangolo](https://github.com/tiangolo).
* 👷 Update GitHub Actions to download and upload artifacts. PR [#936](https://github.com/tiangolo/sqlmodel/pull/936) by [@tiangolo](https://github.com/tiangolo).
* 👷 Tweak CI for test-redistribute, add needed env vars for slim. PR [#929](https://github.com/tiangolo/sqlmodel/pull/929) by [@tiangolo](https://github.com/tiangolo).
## 0.0.18
### Internal
* ✨ Add `sqlmodel-slim` setup. PR [#916](https://github.com/tiangolo/sqlmodel/pull/916) by [@tiangolo](https://github.com/tiangolo).
In the future SQLModel will include the standard default recommended packages, and `sqlmodel-slim` will come without those recommended standard packages and with a group of optional dependencies `sqlmodel-slim[standard]`, equivalent to `sqlmodel`, for those that want to opt out of those packages.
* 🔧 Re-enable MkDocs Material Social plugin. PR [#915](https://github.com/tiangolo/sqlmodel/pull/915) by [@tiangolo](https://github.com/tiangolo).
## 0.0.17
### Refactors

View File

@@ -1,4 +1,3 @@
plugins:
# TODO: Re-enable once this is fixed: https://github.com/squidfunk/mkdocs-material/issues/6983
# social:
social:
typeset:

39
pdm_build.py Normal file
View File

@@ -0,0 +1,39 @@
import os
from typing import Any, Dict, List
from pdm.backend.hooks import Context
TIANGOLO_BUILD_PACKAGE = os.getenv("TIANGOLO_BUILD_PACKAGE", "sqlmodel")
def pdm_build_initialize(context: Context) -> None:
metadata = context.config.metadata
# Get custom config for the current package, from the env var
config: Dict[str, Any] = context.config.data["tool"]["tiangolo"][
"_internal-slim-build"
]["packages"][TIANGOLO_BUILD_PACKAGE]
project_config: Dict[str, Any] = config["project"]
# Get main optional dependencies, extras
optional_dependencies: Dict[str, List[str]] = metadata.get(
"optional-dependencies", {}
)
# Get custom optional dependencies name to always include in this (non-slim) package
include_optional_dependencies: List[str] = config.get(
"include-optional-dependencies", []
)
# Override main [project] configs with custom configs for this package
for key, value in project_config.items():
metadata[key] = value
# Get custom build config for the current package
build_config: Dict[str, Any] = (
config.get("tool", {}).get("pdm", {}).get("build", {})
)
# Override PDM build config with custom build config for this package
for key, value in build_config.items():
context.config.build_config[key] = value
# Get main dependencies
dependencies: List[str] = metadata.get("dependencies", [])
# Add optional dependencies to the default dependencies for this (non-slim) package
for include_optional in include_optional_dependencies:
optional_dependencies_group = optional_dependencies.get(include_optional, [])
dependencies.extend(optional_dependencies_group)

View File

@@ -35,7 +35,7 @@ classifiers = [
]
dependencies = [
"SQLAlchemy >=2.0.0,<2.1.0",
"SQLAlchemy >=2.0.14,<2.1.0",
"pydantic >=1.10.13,<3.0.0",
]
@@ -57,6 +57,18 @@ source-includes = [
"sqlmodel/sql/expression.py.jinja2",
]
[tool.tiangolo._internal-slim-build.packages.sqlmodel-slim.project]
name = "sqlmodel-slim"
[tool.tiangolo._internal-slim-build.packages.sqlmodel]
# include-optional-dependencies = ["standard"]
[tool.tiangolo._internal-slim-build.packages.sqlmodel.project]
optional-dependencies = {}
# [tool.tiangolo._internal-slim-build.packages.sqlmodel.project.scripts]
# sqlmodel = "sqlmodel.cli:main"
[tool.coverage.run]
parallel = true
source = [
@@ -78,7 +90,7 @@ exclude_lines = [
strict = true
[[tool.mypy.overrides]]
module = "sqlmodel.sql.expression"
module = "sqlmodel.sql._expression_select_gen"
warn_unused_ignores = false
[[tool.mypy.overrides]]

View File

@@ -1,2 +1,2 @@
# For mkdocstrings and code generator using templates
black >=22.10,<24.0
black >=22.10

View File

@@ -12,7 +12,7 @@ pillow==10.1.0
# For image processing by Material for MkDocs
cairosvg==2.7.0
mkdocstrings[python]==0.23.0
griffe-typingdoc==0.2.2
# Enable griffe-typingdoc once dropping Python 3.7 and upgrading typing-extensions
# griffe-typingdoc==0.2.5
# For griffe, it formats with black
black==23.3.0
typer == 0.12.3

View File

@@ -3,10 +3,13 @@
pytest >=7.0.1,<8.0.0
coverage[toml] >=6.2,<8.0
mypy ==1.4.1
ruff ==0.2.0
ruff ==0.4.7
# For FastAPI tests
fastapi >=0.103.2
httpx ==0.24.1
# TODO: upgrade when deprecating Python 3.7
dirty-equals ==0.6.0
jinja2 ==3.1.3
# Pin typing-extensions until Python 3.8 is deprecated or the issue with dirty-equals
# is fixed, maybe fixed after dropping Python 3.7 and upgrading dirty-equals
typing-extensions ==4.6.1

View File

@@ -7,8 +7,10 @@ import black
from jinja2 import Template
from pydantic import BaseModel
template_path = Path(__file__).parent.parent / "sqlmodel/sql/expression.py.jinja2"
destiny_path = Path(__file__).parent.parent / "sqlmodel/sql/expression.py"
template_path = (
Path(__file__).parent.parent / "sqlmodel/sql/_expression_select_gen.py.jinja2"
)
destiny_path = Path(__file__).parent.parent / "sqlmodel/sql/_expression_select_gen.py"
number_of_types = 4
@@ -48,7 +50,7 @@ result = template.render(number_of_types=number_of_types, signatures=signatures)
result = (
"# WARNING: do not modify this code, it is generated by "
"expression.py.jinja2\n\n" + result
"_expression_select_gen.py.jinja2\n\n" + result
)
result = black.format_str(result, mode=black.Mode())

View File

@@ -3,7 +3,6 @@
set -e
set -x
CHECK_JINJA=1 python scripts/generate_select.py
coverage run -m pytest tests
coverage combine
coverage report --show-missing

View File

@@ -1,4 +1,4 @@
__version__ = "0.0.17"
__version__ = "0.0.19"
# Re-export from SQLAlchemy
from sqlalchemy.engine import create_engine as create_engine

View File

@@ -72,6 +72,7 @@ def partial_init() -> Generator[None, None, None]:
if IS_PYDANTIC_V2:
from annotated_types import MaxLen
from pydantic import ConfigDict as BaseConfig
from pydantic._internal._fields import PydanticMetadata
from pydantic._internal._model_construction import ModelMetaclass
@@ -201,7 +202,7 @@ if IS_PYDANTIC_V2:
def get_field_metadata(field: Any) -> Any:
for meta in field.metadata:
if isinstance(meta, PydanticMetadata):
if isinstance(meta, (PydanticMetadata, MaxLen)):
return meta
return FakeMetadata()

View File

@@ -43,8 +43,7 @@ class AsyncSession(_AsyncSession):
bind_arguments: Optional[Dict[str, Any]] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
) -> TupleResult[_TSelectParam]:
...
) -> TupleResult[_TSelectParam]: ...
@overload
async def exec(
@@ -56,8 +55,7 @@ class AsyncSession(_AsyncSession):
bind_arguments: Optional[Dict[str, Any]] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
) -> ScalarResult[_TSelectParam]:
...
) -> ScalarResult[_TSelectParam]: ...
async def exec(
self,

View File

@@ -25,7 +25,7 @@ from typing import (
overload,
)
from pydantic import BaseModel
from pydantic import BaseModel, EmailStr
from pydantic.fields import FieldInfo as PydanticFieldInfo
from sqlalchemy import (
Boolean,
@@ -231,8 +231,7 @@ def Field(
sa_column_args: Union[Sequence[Any], UndefinedType] = Undefined,
sa_column_kwargs: Union[Mapping[str, Any], UndefinedType] = Undefined,
schema_extra: Optional[Dict[str, Any]] = None,
) -> Any:
...
) -> Any: ...
@overload
@@ -268,8 +267,7 @@ def Field(
repr: bool = True,
sa_column: Union[Column, UndefinedType] = Undefined, # type: ignore
schema_extra: Optional[Dict[str, Any]] = None,
) -> Any:
...
) -> Any: ...
def Field(
@@ -361,8 +359,7 @@ def Relationship(
link_model: Optional[Any] = None,
sa_relationship_args: Optional[Sequence[Any]] = None,
sa_relationship_kwargs: Optional[Mapping[str, Any]] = None,
) -> Any:
...
) -> Any: ...
@overload
@@ -371,8 +368,7 @@ def Relationship(
back_populates: Optional[str] = None,
link_model: Optional[Any] = None,
sa_relationship: Optional[RelationshipProperty[Any]] = None,
) -> Any:
...
) -> Any: ...
def Relationship(
@@ -574,7 +570,18 @@ def get_sqlalchemy_type(field: Any) -> Any:
# Check enums first as an enum can also be a str, needed by Pydantic/FastAPI
if issubclass(type_, Enum):
return sa_Enum(type_)
if issubclass(type_, str):
if issubclass(
type_,
(
str,
ipaddress.IPv4Address,
ipaddress.IPv4Network,
ipaddress.IPv6Address,
ipaddress.IPv6Network,
Path,
EmailStr,
),
):
max_length = getattr(metadata, "max_length", None)
if max_length:
return AutoString(length=max_length)
@@ -600,16 +607,6 @@ def get_sqlalchemy_type(field: Any) -> Any:
precision=getattr(metadata, "max_digits", None),
scale=getattr(metadata, "decimal_places", None),
)
if issubclass(type_, ipaddress.IPv4Address):
return AutoString
if issubclass(type_, ipaddress.IPv4Network):
return AutoString
if issubclass(type_, ipaddress.IPv6Address):
return AutoString
if issubclass(type_, ipaddress.IPv6Network):
return AutoString
if issubclass(type_, Path):
return AutoString
if issubclass(type_, uuid.UUID):
return GUID
raise ValueError(f"{type_} has no matching SQLAlchemy type")

View File

@@ -35,8 +35,7 @@ class Session(_Session):
bind_arguments: Optional[Dict[str, Any]] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
) -> TupleResult[_TSelectParam]:
...
) -> TupleResult[_TSelectParam]: ...
@overload
def exec(
@@ -48,8 +47,7 @@ class Session(_Session):
bind_arguments: Optional[Dict[str, Any]] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
) -> ScalarResult[_TSelectParam]:
...
) -> ScalarResult[_TSelectParam]: ...
def exec(
self,

View File

@@ -0,0 +1,43 @@
from typing import (
Tuple,
TypeVar,
Union,
)
from sqlalchemy.sql._typing import (
_ColumnExpressionArgument,
)
from sqlalchemy.sql.expression import Select as _Select
from typing_extensions import Self
_T = TypeVar("_T")
# Separate this class in SelectBase, Select, and SelectOfScalar so that they can share
# where and having without having type overlap incompatibility in session.exec().
class SelectBase(_Select[Tuple[_T]]):
inherit_cache = True
def where(self, *whereclause: Union[_ColumnExpressionArgument[bool], bool]) -> Self:
"""Return a new `Select` construct with the given expression added to
its `WHERE` clause, joined to the existing clause via `AND`, if any.
"""
return super().where(*whereclause) # type: ignore[arg-type]
def having(self, *having: Union[_ColumnExpressionArgument[bool], bool]) -> Self:
"""Return a new `Select` construct with the given expression added to
its `HAVING` clause, joined to the existing clause via `AND`, if any.
"""
return super().having(*having) # type: ignore[arg-type]
class Select(SelectBase[_T]):
inherit_cache = True
# This is not comparable to sqlalchemy.sql.selectable.ScalarSelect, that has a different
# purpose. This is the same as a normal SQLAlchemy Select class where there's only one
# entity, so the result will be converted to a scalar by default. This way writing
# for loops on the results will feel natural.
class SelectOfScalar(SelectBase[_T]):
inherit_cache = True

View File

@@ -0,0 +1,367 @@
# WARNING: do not modify this code, it is generated by _expression_select_gen.py.jinja2
from datetime import datetime
from typing import (
Any,
Mapping,
Sequence,
Tuple,
Type,
TypeVar,
Union,
overload,
)
from uuid import UUID
from sqlalchemy import (
Column,
)
from sqlalchemy.sql.elements import (
SQLCoreOperations,
)
from sqlalchemy.sql.roles import TypedColumnsClauseRole
from ._expression_select_cls import Select, SelectOfScalar
_T = TypeVar("_T")
_TCCA = Union[
TypedColumnsClauseRole[_T],
SQLCoreOperations[_T],
Type[_T],
]
# Generated TypeVars start
_TScalar_0 = TypeVar(
"_TScalar_0",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T0 = TypeVar("_T0")
_TScalar_1 = TypeVar(
"_TScalar_1",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T1 = TypeVar("_T1")
_TScalar_2 = TypeVar(
"_TScalar_2",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T2 = TypeVar("_T2")
_TScalar_3 = TypeVar(
"_TScalar_3",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T3 = TypeVar("_T3")
# Generated TypeVars end
@overload
def select(__ent0: _TCCA[_T0]) -> SelectOfScalar[_T0]: ...
@overload
def select(__ent0: _TScalar_0) -> SelectOfScalar[_TScalar_0]: # type: ignore
...
# Generated overloads start
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
) -> Select[Tuple[_T0, _T1]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
) -> Select[Tuple[_T0, _TScalar_1]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
) -> Select[Tuple[_TScalar_0, _T1]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
) -> Select[Tuple[_TScalar_0, _TScalar_1]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
) -> Select[Tuple[_T0, _T1, _T2]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
) -> Select[Tuple[_T0, _T1, _TScalar_2]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
) -> Select[Tuple[_T0, _TScalar_1, _T2]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
entity_2: _TScalar_2,
) -> Select[Tuple[_T0, _TScalar_1, _TScalar_2]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
) -> Select[Tuple[_TScalar_0, _T1, _T2]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
) -> Select[Tuple[_TScalar_0, _T1, _TScalar_2]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
) -> Select[Tuple[_TScalar_0, _TScalar_1, _T2]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
entity_2: _TScalar_2,
) -> Select[Tuple[_TScalar_0, _TScalar_1, _TScalar_2]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _T1, _T2, _T3]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _T1, _T2, _TScalar_3]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _T1, _TScalar_2, _T3]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _T1, _TScalar_2, _TScalar_3]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _TScalar_1, _T2, _T3]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _TScalar_1, _T2, _TScalar_3]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _TScalar_1, _TScalar_2, _T3]]: ...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _TScalar_1, _TScalar_2, _TScalar_3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _T1, _T2, _T3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _T1, _T2, _TScalar_3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _T1, _TScalar_2, _T3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _T1, _TScalar_2, _TScalar_3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _TScalar_1, _T2, _T3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _TScalar_1, _T2, _TScalar_3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _TScalar_1, _TScalar_2, _T3]]: ...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _TScalar_1, _TScalar_2, _TScalar_3]]: ...
# Generated overloads end
def select(*entities: Any) -> Union[Select, SelectOfScalar]: # type: ignore
if len(entities) == 1:
return SelectOfScalar(*entities)
return Select(*entities)

View File

@@ -0,0 +1,84 @@
from datetime import datetime
from typing import (
Any,
Mapping,
Sequence,
Tuple,
Type,
TypeVar,
Union,
overload,
)
from uuid import UUID
from sqlalchemy import (
Column,
)
from sqlalchemy.sql.elements import (
SQLCoreOperations,
)
from sqlalchemy.sql.roles import TypedColumnsClauseRole
from ._expression_select_cls import Select, SelectOfScalar
_T = TypeVar("_T")
_TCCA = Union[
TypedColumnsClauseRole[_T],
SQLCoreOperations[_T],
Type[_T],
]
# Generated TypeVars start
{% for i in range(number_of_types) %}
_TScalar_{{ i }} = TypeVar(
"_TScalar_{{ i }}",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T{{ i }} = TypeVar("_T{{ i }}")
{% endfor %}
# Generated TypeVars end
@overload
def select(__ent0: _TCCA[_T0]) -> SelectOfScalar[_T0]: ...
@overload
def select(__ent0: _TScalar_0) -> SelectOfScalar[_TScalar_0]: # type: ignore
...
# Generated overloads start
{% for signature in signatures %}
@overload
def select( # type: ignore
{% for arg in signature[0] %}{{ arg.name }}: {{ arg.annotation }}, {% endfor %}
) -> Select[Tuple[{%for ret in signature[1] %}{{ ret }} {% if not loop.last %}, {% endif %}{% endfor %}]]: ...
{% endfor %}
# Generated overloads end
def select(*entities: Any) -> Union[Select, SelectOfScalar]: # type: ignore
if len(entities) == 1:
return SelectOfScalar(*entities)
return Select(*entities)

View File

@@ -1,6 +1,3 @@
# WARNING: do not modify this code, it is generated by expression.py.jinja2
from datetime import datetime
from typing import (
Any,
Iterable,
@@ -11,9 +8,7 @@ from typing import (
Type,
TypeVar,
Union,
overload,
)
from uuid import UUID
import sqlalchemy
from sqlalchemy import (
@@ -39,14 +34,15 @@ from sqlalchemy.sql.elements import (
Cast,
CollectionAggregate,
ColumnClause,
SQLCoreOperations,
TryCast,
UnaryExpression,
)
from sqlalchemy.sql.expression import Select as _Select
from sqlalchemy.sql.roles import TypedColumnsClauseRole
from sqlalchemy.sql.type_api import TypeEngine
from typing_extensions import Literal, Self
from typing_extensions import Literal
from ._expression_select_cls import Select as Select
from ._expression_select_cls import SelectOfScalar as SelectOfScalar
from ._expression_select_gen import select as select
_T = TypeVar("_T")
@@ -89,7 +85,7 @@ def between(
upper_bound: Any,
symmetric: bool = False,
) -> BinaryExpression[bool]:
return sqlalchemy.between(expr, lower_bound, upper_bound, symmetric=symmetric) # type: ignore[arg-type]
return sqlalchemy.between(expr, lower_bound, upper_bound, symmetric=symmetric)
def not_(clause: Union[_ColumnExpressionArgument[_T], _T]) -> ColumnElement[_T]:
@@ -110,14 +106,14 @@ def cast(
expression: Union[_ColumnExpressionOrLiteralArgument[Any], Any],
type_: "_TypeEngineArgument[_T]",
) -> Cast[_T]:
return sqlalchemy.cast(expression, type_) # type: ignore[arg-type]
return sqlalchemy.cast(expression, type_)
def try_cast(
expression: Union[_ColumnExpressionOrLiteralArgument[Any], Any],
type_: "_TypeEngineArgument[_T]",
) -> TryCast[_T]:
return sqlalchemy.try_cast(expression, type_) # type: ignore[arg-type]
return sqlalchemy.try_cast(expression, type_)
def desc(
@@ -135,7 +131,7 @@ def bitwise_not(expr: Union[_ColumnExpressionArgument[_T], _T]) -> UnaryExpressi
def extract(field: str, expr: Union[_ColumnExpressionArgument[Any], Any]) -> Extract:
return sqlalchemy.extract(field, expr) # type: ignore[arg-type]
return sqlalchemy.extract(field, expr)
def funcfilter(
@@ -162,7 +158,7 @@ def nulls_last(column: Union[_ColumnExpressionArgument[_T], _T]) -> UnaryExpress
return sqlalchemy.nulls_last(column) # type: ignore[arg-type]
def or_( # type: ignore[empty-body]
def or_(
initial_clause: Union[Literal[False], _ColumnExpressionArgument[bool], bool],
*clauses: Union[_ColumnExpressionArgument[bool], bool],
) -> ColumnElement[bool]:
@@ -190,7 +186,7 @@ def over(
) -> Over[_T]:
return sqlalchemy.over(
element, partition_by=partition_by, order_by=order_by, range_=range_, rows=rows
) # type: ignore[arg-type]
)
def tuple_(
@@ -204,413 +200,13 @@ def type_coerce(
expression: Union[_ColumnExpressionOrLiteralArgument[Any], Any],
type_: "_TypeEngineArgument[_T]",
) -> TypeCoerce[_T]:
return sqlalchemy.type_coerce(expression, type_) # type: ignore[arg-type]
return sqlalchemy.type_coerce(expression, type_)
def within_group(
element: FunctionElement[_T], *order_by: Union[_ColumnExpressionArgument[Any], Any]
) -> WithinGroup[_T]:
return sqlalchemy.within_group(element, *order_by) # type: ignore[arg-type]
# Separate this class in SelectBase, Select, and SelectOfScalar so that they can share
# where and having without having type overlap incompatibility in session.exec().
class SelectBase(_Select[Tuple[_T]]):
inherit_cache = True
def where(self, *whereclause: Union[_ColumnExpressionArgument[bool], bool]) -> Self:
"""Return a new `Select` construct with the given expression added to
its `WHERE` clause, joined to the existing clause via `AND`, if any.
"""
return super().where(*whereclause) # type: ignore[arg-type]
def having(self, *having: Union[_ColumnExpressionArgument[bool], bool]) -> Self:
"""Return a new `Select` construct with the given expression added to
its `HAVING` clause, joined to the existing clause via `AND`, if any.
"""
return super().having(*having) # type: ignore[arg-type]
class Select(SelectBase[_T]):
inherit_cache = True
# This is not comparable to sqlalchemy.sql.selectable.ScalarSelect, that has a different
# purpose. This is the same as a normal SQLAlchemy Select class where there's only one
# entity, so the result will be converted to a scalar by default. This way writing
# for loops on the results will feel natural.
class SelectOfScalar(SelectBase[_T]):
inherit_cache = True
_TCCA = Union[
TypedColumnsClauseRole[_T],
SQLCoreOperations[_T],
Type[_T],
]
# Generated TypeVars start
_TScalar_0 = TypeVar(
"_TScalar_0",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T0 = TypeVar("_T0")
_TScalar_1 = TypeVar(
"_TScalar_1",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T1 = TypeVar("_T1")
_TScalar_2 = TypeVar(
"_TScalar_2",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T2 = TypeVar("_T2")
_TScalar_3 = TypeVar(
"_TScalar_3",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T3 = TypeVar("_T3")
# Generated TypeVars end
@overload
def select(__ent0: _TCCA[_T0]) -> SelectOfScalar[_T0]:
...
@overload
def select(__ent0: _TScalar_0) -> SelectOfScalar[_TScalar_0]: # type: ignore
...
# Generated overloads start
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
) -> Select[Tuple[_T0, _T1]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
) -> Select[Tuple[_T0, _TScalar_1]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
) -> Select[Tuple[_TScalar_0, _T1]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
) -> Select[Tuple[_TScalar_0, _TScalar_1]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
) -> Select[Tuple[_T0, _T1, _T2]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
) -> Select[Tuple[_T0, _T1, _TScalar_2]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
) -> Select[Tuple[_T0, _TScalar_1, _T2]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
entity_2: _TScalar_2,
) -> Select[Tuple[_T0, _TScalar_1, _TScalar_2]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
) -> Select[Tuple[_TScalar_0, _T1, _T2]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
) -> Select[Tuple[_TScalar_0, _T1, _TScalar_2]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
) -> Select[Tuple[_TScalar_0, _TScalar_1, _T2]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
entity_2: _TScalar_2,
) -> Select[Tuple[_TScalar_0, _TScalar_1, _TScalar_2]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _T1, _T2, _T3]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _T1, _T2, _TScalar_3]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _T1, _TScalar_2, _T3]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _T1, _TScalar_2, _TScalar_3]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _TScalar_1, _T2, _T3]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _TScalar_1, _T2, _TScalar_3]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_T0, _TScalar_1, _TScalar_2, _T3]]:
...
@overload
def select( # type: ignore
__ent0: _TCCA[_T0],
entity_1: _TScalar_1,
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_T0, _TScalar_1, _TScalar_2, _TScalar_3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _T1, _T2, _T3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _T1, _T2, _TScalar_3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _T1, _TScalar_2, _T3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
__ent1: _TCCA[_T1],
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _T1, _TScalar_2, _TScalar_3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _TScalar_1, _T2, _T3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
__ent2: _TCCA[_T2],
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _TScalar_1, _T2, _TScalar_3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
entity_2: _TScalar_2,
__ent3: _TCCA[_T3],
) -> Select[Tuple[_TScalar_0, _TScalar_1, _TScalar_2, _T3]]:
...
@overload
def select( # type: ignore
entity_0: _TScalar_0,
entity_1: _TScalar_1,
entity_2: _TScalar_2,
entity_3: _TScalar_3,
) -> Select[Tuple[_TScalar_0, _TScalar_1, _TScalar_2, _TScalar_3]]:
...
# Generated overloads end
def select(*entities: Any) -> Union[Select, SelectOfScalar]: # type: ignore
if len(entities) == 1:
return SelectOfScalar(*entities)
return Select(*entities)
return sqlalchemy.within_group(element, *order_by)
def col(column_expression: _T) -> Mapped[_T]:

View File

@@ -1,309 +0,0 @@
from datetime import datetime
from typing import (
Any,
Iterable,
Mapping,
Optional,
Sequence,
Tuple,
Type,
TypeVar,
Union,
overload,
)
from uuid import UUID
import sqlalchemy
from sqlalchemy import (
Column,
ColumnElement,
Extract,
FunctionElement,
FunctionFilter,
Label,
Over,
TypeCoerce,
WithinGroup,
)
from sqlalchemy.orm import InstrumentedAttribute, Mapped
from sqlalchemy.sql._typing import (
_ColumnExpressionArgument,
_ColumnExpressionOrLiteralArgument,
_ColumnExpressionOrStrLabelArgument,
)
from sqlalchemy.sql.elements import (
BinaryExpression,
Case,
Cast,
CollectionAggregate,
ColumnClause,
SQLCoreOperations,
TryCast,
UnaryExpression,
)
from sqlalchemy.sql.expression import Select as _Select
from sqlalchemy.sql.roles import TypedColumnsClauseRole
from sqlalchemy.sql.type_api import TypeEngine
from typing_extensions import Literal, Self
_T = TypeVar("_T")
_TypeEngineArgument = Union[Type[TypeEngine[_T]], TypeEngine[_T]]
# Redefine operatos that would only take a column expresion to also take the (virtual)
# types of Pydantic models, e.g. str instead of only Mapped[str].
def all_(expr: Union[_ColumnExpressionArgument[_T], _T]) -> CollectionAggregate[bool]:
return sqlalchemy.all_(expr) # type: ignore[arg-type]
def and_(
initial_clause: Union[Literal[True], _ColumnExpressionArgument[bool], bool],
*clauses: Union[_ColumnExpressionArgument[bool], bool],
) -> ColumnElement[bool]:
return sqlalchemy.and_(initial_clause, *clauses) # type: ignore[arg-type]
def any_(expr: Union[_ColumnExpressionArgument[_T], _T]) -> CollectionAggregate[bool]:
return sqlalchemy.any_(expr) # type: ignore[arg-type]
def asc(
column: Union[_ColumnExpressionOrStrLabelArgument[_T], _T],
) -> UnaryExpression[_T]:
return sqlalchemy.asc(column) # type: ignore[arg-type]
def collate(
expression: Union[_ColumnExpressionArgument[str], str], collation: str
) -> BinaryExpression[str]:
return sqlalchemy.collate(expression, collation) # type: ignore[arg-type]
def between(
expr: Union[_ColumnExpressionOrLiteralArgument[_T], _T],
lower_bound: Any,
upper_bound: Any,
symmetric: bool = False,
) -> BinaryExpression[bool]:
return sqlalchemy.between(expr, lower_bound, upper_bound, symmetric=symmetric) # type: ignore[arg-type]
def not_(clause: Union[_ColumnExpressionArgument[_T], _T]) -> ColumnElement[_T]:
return sqlalchemy.not_(clause) # type: ignore[arg-type]
def case(
*whens: Union[
Tuple[Union[_ColumnExpressionArgument[bool], bool], Any], Mapping[Any, Any]
],
value: Optional[Any] = None,
else_: Optional[Any] = None,
) -> Case[Any]:
return sqlalchemy.case(*whens, value=value, else_=else_) # type: ignore[arg-type]
def cast(
expression: Union[_ColumnExpressionOrLiteralArgument[Any], Any],
type_: "_TypeEngineArgument[_T]",
) -> Cast[_T]:
return sqlalchemy.cast(expression, type_) # type: ignore[arg-type]
def try_cast(
expression: Union[_ColumnExpressionOrLiteralArgument[Any], Any],
type_: "_TypeEngineArgument[_T]",
) -> TryCast[_T]:
return sqlalchemy.try_cast(expression, type_) # type: ignore[arg-type]
def desc(
column: Union[_ColumnExpressionOrStrLabelArgument[_T], _T],
) -> UnaryExpression[_T]:
return sqlalchemy.desc(column) # type: ignore[arg-type]
def distinct(expr: Union[_ColumnExpressionArgument[_T], _T]) -> UnaryExpression[_T]:
return sqlalchemy.distinct(expr) # type: ignore[arg-type]
def bitwise_not(expr: Union[_ColumnExpressionArgument[_T], _T]) -> UnaryExpression[_T]:
return sqlalchemy.bitwise_not(expr) # type: ignore[arg-type]
def extract(field: str, expr: Union[_ColumnExpressionArgument[Any], Any]) -> Extract:
return sqlalchemy.extract(field, expr) # type: ignore[arg-type]
def funcfilter(
func: FunctionElement[_T], *criterion: Union[_ColumnExpressionArgument[bool], bool]
) -> FunctionFilter[_T]:
return sqlalchemy.funcfilter(func, *criterion) # type: ignore[arg-type]
def label(
name: str,
element: Union[_ColumnExpressionArgument[_T], _T],
type_: Optional["_TypeEngineArgument[_T]"] = None,
) -> Label[_T]:
return sqlalchemy.label(name, element, type_=type_) # type: ignore[arg-type]
def nulls_first(
column: Union[_ColumnExpressionArgument[_T], _T],
) -> UnaryExpression[_T]:
return sqlalchemy.nulls_first(column) # type: ignore[arg-type]
def nulls_last(column: Union[_ColumnExpressionArgument[_T], _T]) -> UnaryExpression[_T]:
return sqlalchemy.nulls_last(column) # type: ignore[arg-type]
def or_( # type: ignore[empty-body]
initial_clause: Union[Literal[False], _ColumnExpressionArgument[bool], bool],
*clauses: Union[_ColumnExpressionArgument[bool], bool],
) -> ColumnElement[bool]:
return sqlalchemy.or_(initial_clause, *clauses) # type: ignore[arg-type]
def over(
element: FunctionElement[_T],
partition_by: Optional[
Union[
Iterable[Union[_ColumnExpressionArgument[Any], Any]],
_ColumnExpressionArgument[Any],
Any,
]
] = None,
order_by: Optional[
Union[
Iterable[Union[_ColumnExpressionArgument[Any], Any]],
_ColumnExpressionArgument[Any],
Any,
]
] = None,
range_: Optional[Tuple[Optional[int], Optional[int]]] = None,
rows: Optional[Tuple[Optional[int], Optional[int]]] = None,
) -> Over[_T]:
return sqlalchemy.over(
element, partition_by=partition_by, order_by=order_by, range_=range_, rows=rows
) # type: ignore[arg-type]
def tuple_(
*clauses: Union[_ColumnExpressionArgument[Any], Any],
types: Optional[Sequence["_TypeEngineArgument[Any]"]] = None,
) -> Tuple[Any, ...]:
return sqlalchemy.tuple_(*clauses, types=types) # type: ignore[return-value]
def type_coerce(
expression: Union[_ColumnExpressionOrLiteralArgument[Any], Any],
type_: "_TypeEngineArgument[_T]",
) -> TypeCoerce[_T]:
return sqlalchemy.type_coerce(expression, type_) # type: ignore[arg-type]
def within_group(
element: FunctionElement[_T], *order_by: Union[_ColumnExpressionArgument[Any], Any]
) -> WithinGroup[_T]:
return sqlalchemy.within_group(element, *order_by) # type: ignore[arg-type]
# Separate this class in SelectBase, Select, and SelectOfScalar so that they can share
# where and having without having type overlap incompatibility in session.exec().
class SelectBase(_Select[Tuple[_T]]):
inherit_cache = True
def where(self, *whereclause: Union[_ColumnExpressionArgument[bool], bool]) -> Self:
"""Return a new `Select` construct with the given expression added to
its `WHERE` clause, joined to the existing clause via `AND`, if any.
"""
return super().where(*whereclause) # type: ignore[arg-type]
def having(self, *having: Union[_ColumnExpressionArgument[bool], bool]) -> Self:
"""Return a new `Select` construct with the given expression added to
its `HAVING` clause, joined to the existing clause via `AND`, if any.
"""
return super().having(*having) # type: ignore[arg-type]
class Select(SelectBase[_T]):
inherit_cache = True
# This is not comparable to sqlalchemy.sql.selectable.ScalarSelect, that has a different
# purpose. This is the same as a normal SQLAlchemy Select class where there's only one
# entity, so the result will be converted to a scalar by default. This way writing
# for loops on the results will feel natural.
class SelectOfScalar(SelectBase[_T]):
inherit_cache = True
_TCCA = Union[
TypedColumnsClauseRole[_T],
SQLCoreOperations[_T],
Type[_T],
]
# Generated TypeVars start
{% for i in range(number_of_types) %}
_TScalar_{{ i }} = TypeVar(
"_TScalar_{{ i }}",
Column, # type: ignore
Sequence, # type: ignore
Mapping, # type: ignore
UUID,
datetime,
float,
int,
bool,
bytes,
str,
None,
)
_T{{ i }} = TypeVar("_T{{ i }}")
{% endfor %}
# Generated TypeVars end
@overload
def select(__ent0: _TCCA[_T0]) -> SelectOfScalar[_T0]:
...
@overload
def select(__ent0: _TScalar_0) -> SelectOfScalar[_TScalar_0]: # type: ignore
...
# Generated overloads start
{% for signature in signatures %}
@overload
def select( # type: ignore
{% for arg in signature[0] %}{{ arg.name }}: {{ arg.annotation }}, {% endfor %}
) -> Select[Tuple[{%for ret in signature[1] %}{{ ret }} {% if not loop.last %}, {% endif %}{% endfor %}]]:
...
{% endfor %}
# Generated overloads end
def select(*entities: Any) -> Union[Select, SelectOfScalar]: # type: ignore
if len(entities) == 1:
return SelectOfScalar(*entities)
return Select(*entities)
def col(column_expression: _T) -> Mapped[_T]:
if not isinstance(column_expression, (ColumnClause, Column, InstrumentedAttribute)):
raise RuntimeError(f"Not a SQLAlchemy column: {column_expression}")
return column_expression # type: ignore

19
tests/test_select_gen.py Normal file
View File

@@ -0,0 +1,19 @@
import subprocess
import sys
from pathlib import Path
from .conftest import needs_py39
root_path = Path(__file__).parent.parent
@needs_py39
def test_select_gen() -> None:
result = subprocess.run(
[sys.executable, "scripts/generate_select.py"],
env={"CHECK_JINJA": "1"},
check=True,
cwd=root_path,
capture_output=True,
)
print(result.stdout)