| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804380538063807380838093810381138123813381438153816381738183819382038213822382338243825382638273828382938303831383238333834383538363837383838393840384138423843384438453846384738483849385038513852385338543855385638573858385938603861386238633864386538663867386838693870387138723873387438753876387738783879388038813882388338843885388638873888388938903891389238933894389538963897389838993900390139023903390439053906390739083909391039113912391339143915391639173918391939203921392239233924392539263927392839293930393139323933393439353936393739383939394039413942394339443945394639473948394939503951395239533954395539563957395839593960396139623963396439653966396739683969397039713972397339743975397639773978397939803981398239833984398539863987398839893990399139923993399439953996399739983999400040014002400340044005400640074008400940104011401240134014401540164017401840194020402140224023402440254026402740284029403040314032403340344035403640374038403940404041404240434044404540464047404840494050405140524053405440554056405740584059406040614062406340644065406640674068406940704071407240734074407540764077407840794080408140824083408440854086408740884089409040914092409340944095409640974098409941004101410241034104410541064107410841094110411141124113411441154116411741184119412041214122412341244125412641274128412941304131413241334134413541364137413841394140414141424143414441454146414741484149415041514152415341544155415641574158415941604161416241634164416541664167416841694170417141724173417441754176417741784179418041814182418341844185418641874188418941904191419241934194419541964197419841994200420142024203420442054206420742084209421042114212421342144215421642174218421942204221422242234224422542264227422842294230423142324233423442354236423742384239424042414242424342444245424642474248424942504251425242534254425542564257425842594260426142624263426442654266426742684269427042714272427342744275427642774278427942804281428242834284428542864287428842894290429142924293429442954296429742984299430043014302430343044305430643074308430943104311431243134314431543164317431843194320432143224323432443254326432743284329433043314332433343344335433643374338433943404341434243434344434543464347434843494350435143524353435443554356435743584359436043614362436343644365436643674368436943704371437243734374437543764377437843794380438143824383438443854386438743884389439043914392439343944395439643974398439944004401440244034404440544064407440844094410441144124413441444154416441744184419442044214422442344244425442644274428442944304431443244334434443544364437443844394440444144424443444444454446444744484449445044514452445344544455445644574458445944604461446244634464446544664467446844694470447144724473447444754476447744784479448044814482448344844485448644874488448944904491449244934494449544964497449844994500450145024503450445054506450745084509451045114512451345144515451645174518451945204521452245234524452545264527452845294530453145324533453445354536453745384539454045414542454345444545454645474548454945504551455245534554455545564557455845594560456145624563456445654566456745684569457045714572457345744575457645774578457945804581458245834584458545864587458845894590459145924593459445954596459745984599460046014602460346044605460646074608460946104611461246134614461546164617461846194620462146224623462446254626462746284629463046314632463346344635463646374638463946404641464246434644464546464647464846494650465146524653465446554656465746584659466046614662466346644665466646674668466946704671467246734674467546764677467846794680468146824683468446854686468746884689469046914692469346944695469646974698469947004701470247034704470547064707470847094710471147124713471447154716471747184719472047214722472347244725472647274728472947304731473247334734473547364737473847394740474147424743474447454746474747484749475047514752475347544755475647574758475947604761476247634764476547664767476847694770477147724773477447754776477747784779478047814782478347844785478647874788478947904791479247934794479547964797479847994800480148024803480448054806480748084809481048114812481348144815481648174818481948204821482248234824482548264827482848294830483148324833483448354836483748384839484048414842484348444845484648474848484948504851485248534854485548564857485848594860486148624863486448654866486748684869487048714872487348744875487648774878487948804881488248834884488548864887488848894890489148924893489448954896489748984899490049014902490349044905490649074908490949104911491249134914491549164917491849194920492149224923492449254926492749284929493049314932493349344935493649374938493949404941494249434944494549464947494849494950495149524953495449554956495749584959496049614962496349644965496649674968496949704971497249734974497549764977497849794980498149824983498449854986498749884989499049914992499349944995499649974998499950005001500250035004500550065007500850095010501150125013501450155016501750185019502050215022502350245025502650275028502950305031503250335034503550365037503850395040504150425043504450455046504750485049505050515052505350545055505650575058505950605061506250635064506550665067506850695070507150725073507450755076507750785079508050815082508350845085508650875088508950905091509250935094509550965097509850995100510151025103510451055106510751085109511051115112511351145115511651175118511951205121512251235124512551265127512851295130513151325133513451355136513751385139514051415142514351445145514651475148514951505151515251535154515551565157515851595160516151625163516451655166516751685169517051715172517351745175517651775178517951805181518251835184518551865187518851895190519151925193519451955196519751985199520052015202520352045205520652075208520952105211521252135214521552165217521852195220522152225223522452255226522752285229523052315232523352345235523652375238523952405241524252435244524552465247524852495250525152525253525452555256525752585259526052615262526352645265526652675268526952705271527252735274527552765277527852795280528152825283528452855286528752885289529052915292529352945295529652975298529953005301530253035304530553065307530853095310531153125313531453155316531753185319532053215322532353245325532653275328532953305331533253335334533553365337533853395340534153425343534453455346534753485349535053515352535353545355535653575358535953605361536253635364536553665367536853695370537153725373537453755376537753785379538053815382538353845385538653875388538953905391539253935394539553965397539853995400540154025403540454055406540754085409541054115412541354145415541654175418541954205421542254235424542554265427542854295430543154325433543454355436543754385439544054415442544354445445544654475448544954505451545254535454545554565457545854595460546154625463546454655466546754685469547054715472547354745475547654775478547954805481548254835484548554865487548854895490549154925493549454955496549754985499550055015502550355045505550655075508550955105511551255135514551555165517551855195520552155225523552455255526552755285529553055315532553355345535553655375538553955405541554255435544554555465547554855495550555155525553 |
- # sql/elements.py
- # Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
- # <see AUTHORS file>
- #
- # This module is part of SQLAlchemy and is released under
- # the MIT License: https://www.opensource.org/licenses/mit-license.php
- # mypy: allow-untyped-defs, allow-untyped-calls
- """Core SQL expression elements, including :class:`_expression.ClauseElement`,
- :class:`_expression.ColumnElement`, and derived classes.
- """
- from __future__ import annotations
- from decimal import Decimal
- from enum import Enum
- import itertools
- import operator
- import re
- import typing
- from typing import AbstractSet
- from typing import Any
- from typing import Callable
- from typing import cast
- from typing import Dict
- from typing import FrozenSet
- from typing import Generic
- from typing import Iterable
- from typing import Iterator
- from typing import List
- from typing import Mapping
- from typing import Optional
- from typing import overload
- from typing import Sequence
- from typing import Set
- from typing import Tuple as typing_Tuple
- from typing import Type
- from typing import TYPE_CHECKING
- from typing import TypeVar
- from typing import Union
- from . import coercions
- from . import operators
- from . import roles
- from . import traversals
- from . import type_api
- from ._typing import has_schema_attr
- from ._typing import is_named_from_clause
- from ._typing import is_quoted_name
- from ._typing import is_tuple_type
- from .annotation import Annotated
- from .annotation import SupportsWrappingAnnotations
- from .base import _clone
- from .base import _expand_cloned
- from .base import _generative
- from .base import _NoArg
- from .base import Executable
- from .base import Generative
- from .base import HasMemoized
- from .base import Immutable
- from .base import NO_ARG
- from .base import SingletonConstant
- from .cache_key import MemoizedHasCacheKey
- from .cache_key import NO_CACHE
- from .coercions import _document_text_coercion # noqa
- from .operators import ColumnOperators
- from .traversals import HasCopyInternals
- from .visitors import cloned_traverse
- from .visitors import ExternallyTraversible
- from .visitors import InternalTraversal
- from .visitors import traverse
- from .visitors import Visitable
- from .. import exc
- from .. import inspection
- from .. import util
- from ..util import HasMemoized_ro_memoized_attribute
- from ..util import TypingOnly
- from ..util.typing import Literal
- from ..util.typing import ParamSpec
- from ..util.typing import Self
- if typing.TYPE_CHECKING:
- from ._typing import _ByArgument
- from ._typing import _ColumnExpressionArgument
- from ._typing import _ColumnExpressionOrStrLabelArgument
- from ._typing import _HasDialect
- from ._typing import _InfoType
- from ._typing import _PropagateAttrsType
- from ._typing import _TypeEngineArgument
- from .base import _EntityNamespace
- from .base import ColumnSet
- from .cache_key import _CacheKeyTraversalType
- from .cache_key import CacheKey
- from .compiler import Compiled
- from .compiler import SQLCompiler
- from .functions import FunctionElement
- from .operators import OperatorType
- from .schema import Column
- from .schema import DefaultGenerator
- from .schema import FetchedValue
- from .schema import ForeignKey
- from .selectable import _SelectIterable
- from .selectable import FromClause
- from .selectable import NamedFromClause
- from .selectable import TextualSelect
- from .sqltypes import TupleType
- from .type_api import TypeEngine
- from .visitors import _CloneCallableType
- from .visitors import _TraverseInternalsType
- from .visitors import anon_map
- from ..engine import Connection
- from ..engine import Dialect
- from ..engine.interfaces import _CoreMultiExecuteParams
- from ..engine.interfaces import CacheStats
- from ..engine.interfaces import CompiledCacheType
- from ..engine.interfaces import CoreExecuteOptionsParameter
- from ..engine.interfaces import SchemaTranslateMapType
- from ..engine.result import Result
- _NUMERIC = Union[float, Decimal]
- _NUMBER = Union[float, int, Decimal]
- _T = TypeVar("_T", bound="Any")
- _T_co = TypeVar("_T_co", bound=Any, covariant=True)
- _OPT = TypeVar("_OPT", bound="Any")
- _NT = TypeVar("_NT", bound="_NUMERIC")
- _NMT = TypeVar("_NMT", bound="_NUMBER")
- @overload
- def literal(
- value: Any,
- type_: _TypeEngineArgument[_T],
- literal_execute: bool = False,
- ) -> BindParameter[_T]: ...
- @overload
- def literal(
- value: _T,
- type_: None = None,
- literal_execute: bool = False,
- ) -> BindParameter[_T]: ...
- @overload
- def literal(
- value: Any,
- type_: Optional[_TypeEngineArgument[Any]] = None,
- literal_execute: bool = False,
- ) -> BindParameter[Any]: ...
- def literal(
- value: Any,
- type_: Optional[_TypeEngineArgument[Any]] = None,
- literal_execute: bool = False,
- ) -> BindParameter[Any]:
- r"""Return a literal clause, bound to a bind parameter.
- Literal clauses are created automatically when non-
- :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
- etc.) are
- used in a comparison operation with a :class:`_expression.ColumnElement`
- subclass,
- such as a :class:`~sqlalchemy.schema.Column` object. Use this function
- to force the generation of a literal clause, which will be created as a
- :class:`BindParameter` with a bound value.
- :param value: the value to be bound. Can be any Python object supported by
- the underlying DB-API, or is translatable via the given type argument.
- :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
- provide bind-parameter translation for this literal.
- :param literal_execute: optional bool, when True, the SQL engine will
- attempt to render the bound value directly in the SQL statement at
- execution time rather than providing as a parameter value.
- .. versionadded:: 2.0
- """
- return coercions.expect(
- roles.LiteralValueRole,
- value,
- type_=type_,
- literal_execute=literal_execute,
- )
- def literal_column(
- text: str, type_: Optional[_TypeEngineArgument[_T]] = None
- ) -> ColumnClause[_T]:
- r"""Produce a :class:`.ColumnClause` object that has the
- :paramref:`_expression.column.is_literal` flag set to True.
- :func:`_expression.literal_column` is similar to
- :func:`_expression.column`, except that
- it is more often used as a "standalone" column expression that renders
- exactly as stated; while :func:`_expression.column`
- stores a string name that
- will be assumed to be part of a table and may be quoted as such,
- :func:`_expression.literal_column` can be that,
- or any other arbitrary column-oriented
- expression.
- :param text: the text of the expression; can be any SQL expression.
- Quoting rules will not be applied. To specify a column-name expression
- which should be subject to quoting rules, use the :func:`column`
- function.
- :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
- object which will
- provide result-set translation and additional expression semantics for
- this column. If left as ``None`` the type will be :class:`.NullType`.
- .. seealso::
- :func:`_expression.column`
- :func:`_expression.text`
- :ref:`tutorial_select_arbitrary_text`
- """
- return ColumnClause(text, type_=type_, is_literal=True)
- class CompilerElement(Visitable):
- """base class for SQL elements that can be compiled to produce a
- SQL string.
- .. versionadded:: 2.0
- """
- __slots__ = ()
- __visit_name__ = "compiler_element"
- supports_execution = False
- stringify_dialect = "default"
- @util.preload_module("sqlalchemy.engine.default")
- @util.preload_module("sqlalchemy.engine.url")
- def compile(
- self,
- bind: Optional[_HasDialect] = None,
- dialect: Optional[Dialect] = None,
- **kw: Any,
- ) -> Compiled:
- """Compile this SQL expression.
- The return value is a :class:`~.Compiled` object.
- Calling ``str()`` or ``unicode()`` on the returned value will yield a
- string representation of the result. The
- :class:`~.Compiled` object also can return a
- dictionary of bind parameter names and values
- using the ``params`` accessor.
- :param bind: An :class:`.Connection` or :class:`.Engine` which
- can provide a :class:`.Dialect` in order to generate a
- :class:`.Compiled` object. If the ``bind`` and
- ``dialect`` parameters are both omitted, a default SQL compiler
- is used.
- :param column_keys: Used for INSERT and UPDATE statements, a list of
- column names which should be present in the VALUES clause of the
- compiled statement. If ``None``, all columns from the target table
- object are rendered.
- :param dialect: A :class:`.Dialect` instance which can generate
- a :class:`.Compiled` object. This argument takes precedence over
- the ``bind`` argument.
- :param compile_kwargs: optional dictionary of additional parameters
- that will be passed through to the compiler within all "visit"
- methods. This allows any custom flag to be passed through to
- a custom compilation construct, for example. It is also used
- for the case of passing the ``literal_binds`` flag through::
- from sqlalchemy.sql import table, column, select
- t = table("t", column("x"))
- s = select(t).where(t.c.x == 5)
- print(s.compile(compile_kwargs={"literal_binds": True}))
- .. seealso::
- :ref:`faq_sql_expression_string`
- """
- if dialect is None:
- if bind:
- dialect = bind.dialect
- elif self.stringify_dialect == "default":
- dialect = self._default_dialect()
- else:
- url = util.preloaded.engine_url
- dialect = url.URL.create(
- self.stringify_dialect
- ).get_dialect()()
- return self._compiler(dialect, **kw)
- def _default_dialect(self):
- default = util.preloaded.engine_default
- return default.StrCompileDialect()
- def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
- """Return a compiler appropriate for this ClauseElement, given a
- Dialect."""
- if TYPE_CHECKING:
- assert isinstance(self, ClauseElement)
- return dialect.statement_compiler(dialect, self, **kw)
- def __str__(self) -> str:
- return str(self.compile())
- @inspection._self_inspects
- class ClauseElement(
- SupportsWrappingAnnotations,
- MemoizedHasCacheKey,
- HasCopyInternals,
- ExternallyTraversible,
- CompilerElement,
- ):
- """Base class for elements of a programmatically constructed SQL
- expression.
- """
- __visit_name__ = "clause"
- if TYPE_CHECKING:
- @util.memoized_property
- def _propagate_attrs(self) -> _PropagateAttrsType:
- """like annotations, however these propagate outwards liberally
- as SQL constructs are built, and are set up at construction time.
- """
- ...
- else:
- _propagate_attrs = util.EMPTY_DICT
- @util.ro_memoized_property
- def description(self) -> Optional[str]:
- return None
- _is_clone_of: Optional[Self] = None
- is_clause_element = True
- is_selectable = False
- is_dml = False
- _is_column_element = False
- _is_keyed_column_element = False
- _is_table = False
- _gen_static_annotations_cache_key = False
- _is_textual = False
- _is_from_clause = False
- _is_returns_rows = False
- _is_text_clause = False
- _is_from_container = False
- _is_select_container = False
- _is_select_base = False
- _is_select_statement = False
- _is_bind_parameter = False
- _is_clause_list = False
- _is_lambda_element = False
- _is_singleton_constant = False
- _is_immutable = False
- _is_star = False
- @property
- def _order_by_label_element(self) -> Optional[Label[Any]]:
- return None
- _cache_key_traversal: _CacheKeyTraversalType = None
- negation_clause: ColumnElement[bool]
- if typing.TYPE_CHECKING:
- def get_children(
- self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
- ) -> Iterable[ClauseElement]: ...
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return []
- def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
- # usually, self._propagate_attrs is empty here. one case where it's
- # not is a subquery against ORM select, that is then pulled as a
- # property of an aliased class. should all be good
- # assert not self._propagate_attrs
- self._propagate_attrs = util.immutabledict(values)
- return self
- def _default_compiler(self) -> SQLCompiler:
- dialect = self._default_dialect()
- return dialect.statement_compiler(dialect, self) # type: ignore
- def _clone(self, **kw: Any) -> Self:
- """Create a shallow copy of this ClauseElement.
- This method may be used by a generative API. Its also used as
- part of the "deep" copy afforded by a traversal that combines
- the _copy_internals() method.
- """
- skip = self._memoized_keys
- c = self.__class__.__new__(self.__class__)
- if skip:
- # ensure this iteration remains atomic
- c.__dict__ = {
- k: v for k, v in self.__dict__.copy().items() if k not in skip
- }
- else:
- c.__dict__ = self.__dict__.copy()
- # this is a marker that helps to "equate" clauses to each other
- # when a Select returns its list of FROM clauses. the cloning
- # process leaves around a lot of remnants of the previous clause
- # typically in the form of column expressions still attached to the
- # old table.
- cc = self._is_clone_of
- c._is_clone_of = cc if cc is not None else self
- return c
- def _negate_in_binary(self, negated_op, original_op):
- """a hook to allow the right side of a binary expression to respond
- to a negation of the binary expression.
- Used for the special case of expanding bind parameter with IN.
- """
- return self
- def _with_binary_element_type(self, type_):
- """in the context of binary expression, convert the type of this
- object to the one given.
- applies only to :class:`_expression.ColumnElement` classes.
- """
- return self
- @property
- def _constructor(self): # type: ignore[override]
- """return the 'constructor' for this ClauseElement.
- This is for the purposes for creating a new object of
- this type. Usually, its just the element's __class__.
- However, the "Annotated" version of the object overrides
- to return the class of its proxied element.
- """
- return self.__class__
- @HasMemoized.memoized_attribute
- def _cloned_set(self):
- """Return the set consisting all cloned ancestors of this
- ClauseElement.
- Includes this ClauseElement. This accessor tends to be used for
- FromClause objects to identify 'equivalent' FROM clauses, regardless
- of transformative operations.
- """
- s = util.column_set()
- f: Optional[ClauseElement] = self
- # note this creates a cycle, asserted in test_memusage. however,
- # turning this into a plain @property adds tends of thousands of method
- # calls to Core / ORM performance tests, so the small overhead
- # introduced by the relatively small amount of short term cycles
- # produced here is preferable
- while f is not None:
- s.add(f)
- f = f._is_clone_of
- return s
- def _de_clone(self):
- while self._is_clone_of is not None:
- self = self._is_clone_of
- return self
- @util.ro_non_memoized_property
- def entity_namespace(self) -> _EntityNamespace:
- raise AttributeError(
- "This SQL expression has no entity namespace "
- "with which to filter from."
- )
- def __getstate__(self):
- d = self.__dict__.copy()
- d.pop("_is_clone_of", None)
- d.pop("_generate_cache_key", None)
- return d
- def _execute_on_connection(
- self,
- connection: Connection,
- distilled_params: _CoreMultiExecuteParams,
- execution_options: CoreExecuteOptionsParameter,
- ) -> Result[Any]:
- if self.supports_execution:
- if TYPE_CHECKING:
- assert isinstance(self, Executable)
- return connection._execute_clauseelement(
- self, distilled_params, execution_options
- )
- else:
- raise exc.ObjectNotExecutableError(self)
- def _execute_on_scalar(
- self,
- connection: Connection,
- distilled_params: _CoreMultiExecuteParams,
- execution_options: CoreExecuteOptionsParameter,
- ) -> Any:
- """an additional hook for subclasses to provide a different
- implementation for connection.scalar() vs. connection.execute().
- .. versionadded:: 2.0
- """
- return self._execute_on_connection(
- connection, distilled_params, execution_options
- ).scalar()
- def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
- """Return the list of :class:`.BindParameter` objects embedded in the
- object.
- This accomplishes the same purpose as ``visitors.traverse()`` or
- similar would provide, however by making use of the cache key
- it takes advantage of memoization of the key to result in fewer
- net method calls, assuming the statement is also going to be
- executed.
- """
- key = self._generate_cache_key()
- if key is None:
- bindparams: List[BindParameter[Any]] = []
- traverse(self, {}, {"bindparam": bindparams.append})
- return bindparams
- else:
- return key.bindparams
- def unique_params(
- self,
- __optionaldict: Optional[Dict[str, Any]] = None,
- **kwargs: Any,
- ) -> Self:
- """Return a copy with :func:`_expression.bindparam` elements
- replaced.
- Same functionality as :meth:`_expression.ClauseElement.params`,
- except adds `unique=True`
- to affected bind parameters so that multiple statements can be
- used.
- """
- return self._replace_params(True, __optionaldict, kwargs)
- def params(
- self,
- __optionaldict: Optional[Mapping[str, Any]] = None,
- **kwargs: Any,
- ) -> Self:
- """Return a copy with :func:`_expression.bindparam` elements
- replaced.
- Returns a copy of this ClauseElement with
- :func:`_expression.bindparam`
- elements replaced with values taken from the given dictionary::
- >>> clause = column("x") + bindparam("foo")
- >>> print(clause.compile().params)
- {'foo':None}
- >>> print(clause.params({"foo": 7}).compile().params)
- {'foo':7}
- """
- return self._replace_params(False, __optionaldict, kwargs)
- def _replace_params(
- self,
- unique: bool,
- optionaldict: Optional[Mapping[str, Any]],
- kwargs: Dict[str, Any],
- ) -> Self:
- if optionaldict:
- kwargs.update(optionaldict)
- def visit_bindparam(bind: BindParameter[Any]) -> None:
- if bind.key in kwargs:
- bind.value = kwargs[bind.key]
- bind.required = False
- if unique:
- bind._convert_to_unique()
- return cloned_traverse(
- self,
- {"maintain_key": True, "detect_subquery_cols": True},
- {"bindparam": visit_bindparam},
- )
- def compare(self, other: ClauseElement, **kw: Any) -> bool:
- r"""Compare this :class:`_expression.ClauseElement` to
- the given :class:`_expression.ClauseElement`.
- Subclasses should override the default behavior, which is a
- straight identity comparison.
- \**kw are arguments consumed by subclass ``compare()`` methods and
- may be used to modify the criteria for comparison
- (see :class:`_expression.ColumnElement`).
- """
- return traversals.compare(self, other, **kw)
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> ClauseElement:
- """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
- This method is overridden by subclasses to return a "grouping"
- construct, i.e. parenthesis. In particular it's used by "binary"
- expressions to provide a grouping around themselves when placed into a
- larger expression, as well as by :func:`_expression.select`
- constructs when placed into the FROM clause of another
- :func:`_expression.select`. (Note that subqueries should be
- normally created using the :meth:`_expression.Select.alias` method,
- as many
- platforms require nested SELECT statements to be named).
- As expressions are composed together, the application of
- :meth:`self_group` is automatic - end-user code should never
- need to use this method directly. Note that SQLAlchemy's
- clause constructs take operator precedence into account -
- so parenthesis might not be needed, for example, in
- an expression like ``x OR (y AND z)`` - AND takes precedence
- over OR.
- The base :meth:`self_group` method of
- :class:`_expression.ClauseElement`
- just returns self.
- """
- return self
- def _ungroup(self) -> ClauseElement:
- """Return this :class:`_expression.ClauseElement`
- without any groupings.
- """
- return self
- def _compile_w_cache(
- self,
- dialect: Dialect,
- *,
- compiled_cache: Optional[CompiledCacheType],
- column_keys: List[str],
- for_executemany: bool = False,
- schema_translate_map: Optional[SchemaTranslateMapType] = None,
- **kw: Any,
- ) -> typing_Tuple[
- Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
- ]:
- elem_cache_key: Optional[CacheKey]
- if compiled_cache is not None and dialect._supports_statement_cache:
- elem_cache_key = self._generate_cache_key()
- else:
- elem_cache_key = None
- extracted_params: Optional[Sequence[BindParameter[Any]]]
- if elem_cache_key is not None:
- if TYPE_CHECKING:
- assert compiled_cache is not None
- cache_key, extracted_params = elem_cache_key
- key = (
- dialect,
- cache_key,
- tuple(column_keys),
- bool(schema_translate_map),
- for_executemany,
- )
- compiled_sql = compiled_cache.get(key)
- if compiled_sql is None:
- cache_hit = dialect.CACHE_MISS
- compiled_sql = self._compiler(
- dialect,
- cache_key=elem_cache_key,
- column_keys=column_keys,
- for_executemany=for_executemany,
- schema_translate_map=schema_translate_map,
- **kw,
- )
- compiled_cache[key] = compiled_sql
- else:
- cache_hit = dialect.CACHE_HIT
- else:
- extracted_params = None
- compiled_sql = self._compiler(
- dialect,
- cache_key=elem_cache_key,
- column_keys=column_keys,
- for_executemany=for_executemany,
- schema_translate_map=schema_translate_map,
- **kw,
- )
- if not dialect._supports_statement_cache:
- cache_hit = dialect.NO_DIALECT_SUPPORT
- elif compiled_cache is None:
- cache_hit = dialect.CACHING_DISABLED
- else:
- cache_hit = dialect.NO_CACHE_KEY
- return compiled_sql, extracted_params, cache_hit
- def __invert__(self):
- # undocumented element currently used by the ORM for
- # relationship.contains()
- if hasattr(self, "negation_clause"):
- return self.negation_clause
- else:
- return self._negate()
- def _negate(self) -> ClauseElement:
- # TODO: this code is uncovered and in all likelihood is not included
- # in any codepath. So this should raise NotImplementedError in 2.1
- grouped = self.self_group(against=operators.inv)
- assert isinstance(grouped, ColumnElement)
- return UnaryExpression(grouped, operator=operators.inv)
- def __bool__(self):
- raise TypeError("Boolean value of this clause is not defined")
- def __repr__(self):
- friendly = self.description
- if friendly is None:
- return object.__repr__(self)
- else:
- return "<%s.%s at 0x%x; %s>" % (
- self.__module__,
- self.__class__.__name__,
- id(self),
- friendly,
- )
- class DQLDMLClauseElement(ClauseElement):
- """represents a :class:`.ClauseElement` that compiles to a DQL or DML
- expression, not DDL.
- .. versionadded:: 2.0
- """
- if typing.TYPE_CHECKING:
- def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
- """Return a compiler appropriate for this ClauseElement, given a
- Dialect."""
- ...
- def compile( # noqa: A001
- self,
- bind: Optional[_HasDialect] = None,
- dialect: Optional[Dialect] = None,
- **kw: Any,
- ) -> SQLCompiler: ...
- class CompilerColumnElement(
- roles.DMLColumnRole,
- roles.DDLConstraintColumnRole,
- roles.ColumnsClauseRole,
- CompilerElement,
- ):
- """A compiler-only column element used for ad-hoc string compilations.
- .. versionadded:: 2.0
- """
- __slots__ = ()
- _propagate_attrs = util.EMPTY_DICT
- _is_collection_aggregate = False
- # SQLCoreOperations should be suiting the ExpressionElementRole
- # and ColumnsClauseRole. however the MRO issues become too elaborate
- # at the moment.
- class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
- __slots__ = ()
- # annotations for comparison methods
- # these are from operators->Operators / ColumnOperators,
- # redefined with the specific types returned by ColumnElement hierarchies
- if typing.TYPE_CHECKING:
- @util.non_memoized_property
- def _propagate_attrs(self) -> _PropagateAttrsType: ...
- def operate(
- self, op: OperatorType, *other: Any, **kwargs: Any
- ) -> ColumnElement[Any]: ...
- def reverse_operate(
- self, op: OperatorType, other: Any, **kwargs: Any
- ) -> ColumnElement[Any]: ...
- @overload
- def op(
- self,
- opstring: str,
- precedence: int = ...,
- is_comparison: bool = ...,
- *,
- return_type: _TypeEngineArgument[_OPT],
- python_impl: Optional[Callable[..., Any]] = None,
- ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
- @overload
- def op(
- self,
- opstring: str,
- precedence: int = ...,
- is_comparison: bool = ...,
- return_type: Optional[_TypeEngineArgument[Any]] = ...,
- python_impl: Optional[Callable[..., Any]] = ...,
- ) -> Callable[[Any], BinaryExpression[Any]]: ...
- def op(
- self,
- opstring: str,
- precedence: int = 0,
- is_comparison: bool = False,
- return_type: Optional[_TypeEngineArgument[Any]] = None,
- python_impl: Optional[Callable[..., Any]] = None,
- ) -> Callable[[Any], BinaryExpression[Any]]: ...
- def bool_op(
- self,
- opstring: str,
- precedence: int = 0,
- python_impl: Optional[Callable[..., Any]] = None,
- ) -> Callable[[Any], BinaryExpression[bool]]: ...
- def __and__(self, other: Any) -> BooleanClauseList: ...
- def __or__(self, other: Any) -> BooleanClauseList: ...
- def __invert__(self) -> ColumnElement[_T_co]: ...
- def __lt__(self, other: Any) -> ColumnElement[bool]: ...
- def __le__(self, other: Any) -> ColumnElement[bool]: ...
- # declare also that this class has an hash method otherwise
- # it may be assumed to be None by type checkers since the
- # object defines __eq__ and python sets it to None in that case:
- # https://docs.python.org/3/reference/datamodel.html#object.__hash__
- def __hash__(self) -> int: ...
- def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
- ...
- def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
- ...
- def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
- def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
- def __gt__(self, other: Any) -> ColumnElement[bool]: ...
- def __ge__(self, other: Any) -> ColumnElement[bool]: ...
- def __neg__(self) -> UnaryExpression[_T_co]: ...
- def __contains__(self, other: Any) -> ColumnElement[bool]: ...
- def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
- @overload
- def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
- @overload
- def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
- def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
- @overload
- def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
- def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
- @overload
- def concat(self, other: Any) -> ColumnElement[Any]: ...
- def concat(self, other: Any) -> ColumnElement[Any]: ...
- def like(
- self, other: Any, escape: Optional[str] = None
- ) -> BinaryExpression[bool]: ...
- def ilike(
- self, other: Any, escape: Optional[str] = None
- ) -> BinaryExpression[bool]: ...
- def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
- def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
- def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
- def bitwise_not(self) -> UnaryExpression[_T_co]: ...
- def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
- def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
- def in_(
- self,
- other: Union[
- Iterable[Any], BindParameter[Any], roles.InElementRole
- ],
- ) -> BinaryExpression[bool]: ...
- def not_in(
- self,
- other: Union[
- Iterable[Any], BindParameter[Any], roles.InElementRole
- ],
- ) -> BinaryExpression[bool]: ...
- def notin_(
- self,
- other: Union[
- Iterable[Any], BindParameter[Any], roles.InElementRole
- ],
- ) -> BinaryExpression[bool]: ...
- def not_like(
- self, other: Any, escape: Optional[str] = None
- ) -> BinaryExpression[bool]: ...
- def notlike(
- self, other: Any, escape: Optional[str] = None
- ) -> BinaryExpression[bool]: ...
- def not_ilike(
- self, other: Any, escape: Optional[str] = None
- ) -> BinaryExpression[bool]: ...
- def notilike(
- self, other: Any, escape: Optional[str] = None
- ) -> BinaryExpression[bool]: ...
- def is_(self, other: Any) -> BinaryExpression[bool]: ...
- def is_not(self, other: Any) -> BinaryExpression[bool]: ...
- def isnot(self, other: Any) -> BinaryExpression[bool]: ...
- def startswith(
- self,
- other: Any,
- escape: Optional[str] = None,
- autoescape: bool = False,
- ) -> ColumnElement[bool]: ...
- def istartswith(
- self,
- other: Any,
- escape: Optional[str] = None,
- autoescape: bool = False,
- ) -> ColumnElement[bool]: ...
- def endswith(
- self,
- other: Any,
- escape: Optional[str] = None,
- autoescape: bool = False,
- ) -> ColumnElement[bool]: ...
- def iendswith(
- self,
- other: Any,
- escape: Optional[str] = None,
- autoescape: bool = False,
- ) -> ColumnElement[bool]: ...
- def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
- def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
- def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
- def regexp_match(
- self, pattern: Any, flags: Optional[str] = None
- ) -> ColumnElement[bool]: ...
- def regexp_replace(
- self, pattern: Any, replacement: Any, flags: Optional[str] = None
- ) -> ColumnElement[str]: ...
- def desc(self) -> UnaryExpression[_T_co]: ...
- def asc(self) -> UnaryExpression[_T_co]: ...
- def nulls_first(self) -> UnaryExpression[_T_co]: ...
- def nullsfirst(self) -> UnaryExpression[_T_co]: ...
- def nulls_last(self) -> UnaryExpression[_T_co]: ...
- def nullslast(self) -> UnaryExpression[_T_co]: ...
- def collate(self, collation: str) -> CollationClause: ...
- def between(
- self, cleft: Any, cright: Any, symmetric: bool = False
- ) -> BinaryExpression[bool]: ...
- def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
- def any_(self) -> CollectionAggregate[Any]: ...
- def all_(self) -> CollectionAggregate[Any]: ...
- # numeric overloads. These need more tweaking
- # in particular they all need to have a variant for Optiona[_T]
- # because Optional only applies to the data side, not the expression
- # side
- @overload
- def __add__(
- self: _SQO[_NMT],
- other: Any,
- ) -> ColumnElement[_NMT]: ...
- @overload
- def __add__(
- self: _SQO[str],
- other: Any,
- ) -> ColumnElement[str]: ...
- @overload
- def __add__(self, other: Any) -> ColumnElement[Any]: ...
- def __add__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
- @overload
- def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
- def __radd__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __sub__(
- self: _SQO[_NMT],
- other: Any,
- ) -> ColumnElement[_NMT]: ...
- @overload
- def __sub__(self, other: Any) -> ColumnElement[Any]: ...
- def __sub__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __rsub__(
- self: _SQO[_NMT],
- other: Any,
- ) -> ColumnElement[_NMT]: ...
- @overload
- def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
- def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __mul__(
- self: _SQO[_NMT],
- other: Any,
- ) -> ColumnElement[_NMT]: ...
- @overload
- def __mul__(self, other: Any) -> ColumnElement[Any]: ...
- def __mul__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __rmul__(
- self: _SQO[_NMT],
- other: Any,
- ) -> ColumnElement[_NMT]: ...
- @overload
- def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
- def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
- @overload
- def __mod__(self, other: Any) -> ColumnElement[Any]: ...
- def __mod__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
- @overload
- def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
- def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __truediv__(
- self: _SQO[int], other: Any
- ) -> ColumnElement[_NUMERIC]: ...
- @overload
- def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
- @overload
- def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
- def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __rtruediv__(
- self: _SQO[_NMT], other: Any
- ) -> ColumnElement[_NUMERIC]: ...
- @overload
- def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
- def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __floordiv__(
- self: _SQO[_NMT], other: Any
- ) -> ColumnElement[_NMT]: ...
- @overload
- def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
- def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
- @overload
- def __rfloordiv__(
- self: _SQO[_NMT], other: Any
- ) -> ColumnElement[_NMT]: ...
- @overload
- def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
- def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
- class SQLColumnExpression(
- SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
- ):
- """A type that may be used to indicate any SQL column element or object
- that acts in place of one.
- :class:`.SQLColumnExpression` is a base of
- :class:`.ColumnElement`, as well as within the bases of ORM elements
- such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
- typing to indicate arguments or return values that should behave
- as column expressions.
- .. versionadded:: 2.0.0b4
- """
- __slots__ = ()
- _SQO = SQLCoreOperations
- class ColumnElement(
- roles.ColumnArgumentOrKeyRole,
- roles.StatementOptionRole,
- roles.WhereHavingRole,
- roles.BinaryElementRole[_T],
- roles.OrderByRole,
- roles.ColumnsClauseRole,
- roles.LimitOffsetRole,
- roles.DMLColumnRole,
- roles.DDLConstraintColumnRole,
- roles.DDLExpressionRole,
- SQLColumnExpression[_T],
- DQLDMLClauseElement,
- ):
- """Represent a column-oriented SQL expression suitable for usage in the
- "columns" clause, WHERE clause etc. of a statement.
- While the most familiar kind of :class:`_expression.ColumnElement` is the
- :class:`_schema.Column` object, :class:`_expression.ColumnElement`
- serves as the basis
- for any unit that may be present in a SQL expression, including
- the expressions themselves, SQL functions, bound parameters,
- literal expressions, keywords such as ``NULL``, etc.
- :class:`_expression.ColumnElement`
- is the ultimate base class for all such elements.
- A wide variety of SQLAlchemy Core functions work at the SQL expression
- level, and are intended to accept instances of
- :class:`_expression.ColumnElement` as
- arguments. These functions will typically document that they accept a
- "SQL expression" as an argument. What this means in terms of SQLAlchemy
- usually refers to an input which is either already in the form of a
- :class:`_expression.ColumnElement` object,
- or a value which can be **coerced** into
- one. The coercion rules followed by most, but not all, SQLAlchemy Core
- functions with regards to SQL expressions are as follows:
- * a literal Python value, such as a string, integer or floating
- point value, boolean, datetime, ``Decimal`` object, or virtually
- any other Python object, will be coerced into a "literal bound
- value". This generally means that a :func:`.bindparam` will be
- produced featuring the given value embedded into the construct; the
- resulting :class:`.BindParameter` object is an instance of
- :class:`_expression.ColumnElement`.
- The Python value will ultimately be sent
- to the DBAPI at execution time as a parameterized argument to the
- ``execute()`` or ``executemany()`` methods, after SQLAlchemy
- type-specific converters (e.g. those provided by any associated
- :class:`.TypeEngine` objects) are applied to the value.
- * any special object value, typically ORM-level constructs, which
- feature an accessor called ``__clause_element__()``. The Core
- expression system looks for this method when an object of otherwise
- unknown type is passed to a function that is looking to coerce the
- argument into a :class:`_expression.ColumnElement` and sometimes a
- :class:`_expression.SelectBase` expression.
- It is used within the ORM to
- convert from ORM-specific objects like mapped classes and
- mapped attributes into Core expression objects.
- * The Python ``None`` value is typically interpreted as ``NULL``,
- which in SQLAlchemy Core produces an instance of :func:`.null`.
- A :class:`_expression.ColumnElement` provides the ability to generate new
- :class:`_expression.ColumnElement`
- objects using Python expressions. This means that Python operators
- such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
- and allow the instantiation of further :class:`_expression.ColumnElement`
- instances
- which are composed from other, more fundamental
- :class:`_expression.ColumnElement`
- objects. For example, two :class:`.ColumnClause` objects can be added
- together with the addition operator ``+`` to produce
- a :class:`.BinaryExpression`.
- Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
- of :class:`_expression.ColumnElement`:
- .. sourcecode:: pycon+sql
- >>> from sqlalchemy.sql import column
- >>> column("a") + column("b")
- <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
- >>> print(column("a") + column("b"))
- {printsql}a + b
- .. seealso::
- :class:`_schema.Column`
- :func:`_expression.column`
- """
- __visit_name__ = "column_element"
- primary_key: bool = False
- _is_clone_of: Optional[ColumnElement[_T]]
- _is_column_element = True
- _insert_sentinel: bool = False
- _omit_from_statements = False
- _is_collection_aggregate = False
- foreign_keys: AbstractSet[ForeignKey] = frozenset()
- @util.memoized_property
- def _proxies(self) -> List[ColumnElement[Any]]:
- return []
- @util.non_memoized_property
- def _tq_label(self) -> Optional[str]:
- """The named label that can be used to target
- this column in a result set in a "table qualified" context.
- This label is almost always the label used when
- rendering <expr> AS <label> in a SELECT statement when using
- the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
- legacy ORM ``Query`` object uses as well.
- For a regular Column bound to a Table, this is typically the label
- <tablename>_<columnname>. For other constructs, different rules
- may apply, such as anonymized labels and others.
- .. versionchanged:: 1.4.21 renamed from ``._label``
- """
- return None
- key: Optional[str] = None
- """The 'key' that in some circumstances refers to this object in a
- Python namespace.
- This typically refers to the "key" of the column as present in the
- ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
- return a :class:`_schema.Column` with a ``.key`` of "somekey".
- """
- @HasMemoized.memoized_attribute
- def _tq_key_label(self) -> Optional[str]:
- """A label-based version of 'key' that in some circumstances refers
- to this object in a Python namespace.
- _tq_key_label comes into play when a select() statement is constructed
- with apply_labels(); in this case, all Column objects in the ``.c``
- collection are rendered as <tablename>_<columnname> in SQL; this is
- essentially the value of ._label. But to locate those columns in the
- ``.c`` collection, the name is along the lines of <tablename>_<key>;
- that's the typical value of .key_label.
- .. versionchanged:: 1.4.21 renamed from ``._key_label``
- """
- return self._proxy_key
- @property
- def _key_label(self) -> Optional[str]:
- """legacy; renamed to _tq_key_label"""
- return self._tq_key_label
- @property
- def _label(self) -> Optional[str]:
- """legacy; renamed to _tq_label"""
- return self._tq_label
- @property
- def _non_anon_label(self) -> Optional[str]:
- """the 'name' that naturally applies this element when rendered in
- SQL.
- Concretely, this is the "name" of a column or a label in a
- SELECT statement; ``<columnname>`` and ``<labelname>`` below:
- .. sourcecode:: sql
- SELECT <columnmame> FROM table
- SELECT column AS <labelname> FROM table
- Above, the two names noted will be what's present in the DBAPI
- ``cursor.description`` as the names.
- If this attribute returns ``None``, it means that the SQL element as
- written does not have a 100% fully predictable "name" that would appear
- in the ``cursor.description``. Examples include SQL functions, CAST
- functions, etc. While such things do return names in
- ``cursor.description``, they are only predictable on a
- database-specific basis; e.g. an expression like ``MAX(table.col)`` may
- appear as the string ``max`` on one database (like PostgreSQL) or may
- appear as the whole expression ``max(table.col)`` on SQLite.
- The default implementation looks for a ``.name`` attribute on the
- object, as has been the precedent established in SQLAlchemy for many
- years. An exception is made on the ``FunctionElement`` subclass
- so that the return value is always ``None``.
- .. versionadded:: 1.4.21
- """
- return getattr(self, "name", None)
- _render_label_in_columns_clause = True
- """A flag used by select._columns_plus_names that helps to determine
- we are actually going to render in terms of "SELECT <col> AS <label>".
- This flag can be returned as False for some Column objects that want
- to be rendered as simple "SELECT <col>"; typically columns that don't have
- any parent table and are named the same as what the label would be
- in any case.
- """
- _allow_label_resolve = True
- """A flag that can be flipped to prevent a column from being resolvable
- by string label name.
- The joined eager loader strategy in the ORM uses this, for example.
- """
- _is_implicitly_boolean = False
- _alt_names: Sequence[str] = ()
- if TYPE_CHECKING:
- def _ungroup(self) -> ColumnElement[_T]: ...
- @overload
- def self_group(self, against: None = None) -> ColumnElement[_T]: ...
- @overload
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> ColumnElement[Any]: ...
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> ColumnElement[Any]:
- if (
- against in (operators.and_, operators.or_, operators._asbool)
- and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
- ):
- return AsBoolean(self, operators.is_true, operators.is_false)
- elif against in (operators.any_op, operators.all_op):
- return Grouping(self)
- else:
- return self
- @overload
- def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
- @overload
- def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
- def _negate(self) -> ColumnElement[Any]:
- if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
- return AsBoolean(self, operators.is_false, operators.is_true)
- else:
- grouped = self.self_group(against=operators.inv)
- assert isinstance(grouped, ColumnElement)
- return UnaryExpression(
- grouped,
- operator=operators.inv,
- )
- type: TypeEngine[_T]
- if not TYPE_CHECKING:
- @util.memoized_property
- def type(self) -> TypeEngine[_T]: # noqa: A001
- # used for delayed setup of
- # type_api
- return type_api.NULLTYPE
- @HasMemoized.memoized_attribute
- def comparator(self) -> TypeEngine.Comparator[_T]:
- try:
- comparator_factory = self.type.comparator_factory
- except AttributeError as err:
- raise TypeError(
- "Object %r associated with '.type' attribute "
- "is not a TypeEngine class or object" % self.type
- ) from err
- else:
- return comparator_factory(self)
- def __setstate__(self, state):
- self.__dict__.update(state)
- def __getattr__(self, key: str) -> Any:
- try:
- return getattr(self.comparator, key)
- except AttributeError as err:
- raise AttributeError(
- "Neither %r object nor %r object has an attribute %r"
- % (
- type(self).__name__,
- type(self.comparator).__name__,
- key,
- )
- ) from err
- def operate(
- self,
- op: operators.OperatorType,
- *other: Any,
- **kwargs: Any,
- ) -> ColumnElement[Any]:
- return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
- def reverse_operate(
- self, op: operators.OperatorType, other: Any, **kwargs: Any
- ) -> ColumnElement[Any]:
- return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
- def _bind_param(
- self,
- operator: operators.OperatorType,
- obj: Any,
- type_: Optional[TypeEngine[_T]] = None,
- expanding: bool = False,
- ) -> BindParameter[_T]:
- return BindParameter(
- None,
- obj,
- _compared_to_operator=operator,
- type_=type_,
- _compared_to_type=self.type,
- unique=True,
- expanding=expanding,
- )
- @property
- def expression(self) -> ColumnElement[Any]:
- """Return a column expression.
- Part of the inspection interface; returns self.
- """
- return self
- @property
- def _select_iterable(self) -> _SelectIterable:
- return (self,)
- @util.memoized_property
- def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
- return frozenset(c for c in self.proxy_set if not c._proxies)
- @util.memoized_property
- def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
- """set of all columns we are proxying
- as of 2.0 this is explicitly deannotated columns. previously it was
- effectively deannotated columns but wasn't enforced. annotated
- columns should basically not go into sets if at all possible because
- their hashing behavior is very non-performant.
- """
- return frozenset([self._deannotate()]).union(
- itertools.chain(*[c.proxy_set for c in self._proxies])
- )
- @util.memoized_property
- def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
- return frozenset(_expand_cloned(self.proxy_set))
- def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
- """An 'uncached' version of proxy set.
- This list includes annotated columns which perform very poorly in
- set operations.
- """
- return [self] + list(
- itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
- )
- def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
- """Return True if the given :class:`_expression.ColumnElement`
- has a common ancestor to this :class:`_expression.ColumnElement`."""
- return bool(self.proxy_set.intersection(othercolumn.proxy_set))
- def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
- """Return True if the given column element compares to this one
- when targeting within a result row."""
- return (
- hasattr(other, "name")
- and hasattr(self, "name")
- and other.name == self.name
- )
- @HasMemoized.memoized_attribute
- def _proxy_key(self) -> Optional[str]:
- if self._annotations and "proxy_key" in self._annotations:
- return cast(str, self._annotations["proxy_key"])
- name = self.key
- if not name:
- # there's a bit of a seeming contradiction which is that the
- # "_non_anon_label" of a column can in fact be an
- # "_anonymous_label"; this is when it's on a column that is
- # proxying for an anonymous expression in a subquery.
- name = self._non_anon_label
- if isinstance(name, _anonymous_label):
- return None
- else:
- return name
- @HasMemoized.memoized_attribute
- def _expression_label(self) -> Optional[str]:
- """a suggested label to use in the case that the column has no name,
- which should be used if possible as the explicit 'AS <label>'
- where this expression would normally have an anon label.
- this is essentially mostly what _proxy_key does except it returns
- None if the column has a normal name that can be used.
- """
- if getattr(self, "name", None) is not None:
- return None
- elif self._annotations and "proxy_key" in self._annotations:
- return cast(str, self._annotations["proxy_key"])
- else:
- return None
- def _make_proxy(
- self,
- selectable: FromClause,
- *,
- primary_key: ColumnSet,
- foreign_keys: Set[KeyedColumnElement[Any]],
- name: Optional[str] = None,
- key: Optional[str] = None,
- name_is_truncatable: bool = False,
- compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
- **kw: Any,
- ) -> typing_Tuple[str, ColumnClause[_T]]:
- """Create a new :class:`_expression.ColumnElement` representing this
- :class:`_expression.ColumnElement` as it appears in the select list of
- a descending selectable.
- """
- if name is None:
- name = self._anon_name_label
- if key is None:
- key = self._proxy_key
- else:
- key = name
- assert key is not None
- co: ColumnClause[_T] = ColumnClause(
- (
- coercions.expect(roles.TruncatedLabelRole, name)
- if name_is_truncatable
- else name
- ),
- type_=getattr(self, "type", None),
- _selectable=selectable,
- )
- co._propagate_attrs = selectable._propagate_attrs
- if compound_select_cols:
- co._proxies = list(compound_select_cols)
- else:
- co._proxies = [self]
- if selectable._is_clone_of is not None:
- co._is_clone_of = selectable._is_clone_of.columns.get(key)
- return key, co
- def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
- """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
- This is a shortcut to the :func:`_expression.cast` function.
- .. seealso::
- :ref:`tutorial_casts`
- :func:`_expression.cast`
- :func:`_expression.type_coerce`
- """
- return Cast(self, type_)
- def label(self, name: Optional[str]) -> Label[_T]:
- """Produce a column label, i.e. ``<columnname> AS <name>``.
- This is a shortcut to the :func:`_expression.label` function.
- If 'name' is ``None``, an anonymous label name will be generated.
- """
- return Label(name, self, self.type)
- def _anon_label(
- self, seed: Optional[str], add_hash: Optional[int] = None
- ) -> _anonymous_label:
- while self._is_clone_of is not None:
- self = self._is_clone_of
- # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
- # as the identifier, because a column and its annotated version are
- # the same thing in a SQL statement
- hash_value = hash(self)
- if add_hash:
- # this path is used for disambiguating anon labels that would
- # otherwise be the same name for the same element repeated.
- # an additional numeric value is factored in for each label.
- # shift hash(self) (which is id(self), typically 8 byte integer)
- # 16 bits leftward. fill extra add_hash on right
- assert add_hash < (2 << 15)
- assert seed
- hash_value = (hash_value << 16) | add_hash
- # extra underscore is added for labels with extra hash
- # values, to isolate the "deduped anon" namespace from the
- # regular namespace. eliminates chance of these
- # manufactured hash values overlapping with regular ones for some
- # undefined python interpreter
- seed = seed + "_"
- if isinstance(seed, _anonymous_label):
- return _anonymous_label.safe_construct(
- hash_value, "", enclosing_label=seed
- )
- return _anonymous_label.safe_construct(hash_value, seed or "anon")
- @util.memoized_property
- def _anon_name_label(self) -> str:
- """Provides a constant 'anonymous label' for this ColumnElement.
- This is a label() expression which will be named at compile time.
- The same label() is returned each time ``anon_label`` is called so
- that expressions can reference ``anon_label`` multiple times,
- producing the same label name at compile time.
- The compiler uses this function automatically at compile time
- for expressions that are known to be 'unnamed' like binary
- expressions and function calls.
- .. versionchanged:: 1.4.9 - this attribute was not intended to be
- public and is renamed to _anon_name_label. anon_name exists
- for backwards compat
- """
- name = getattr(self, "name", None)
- return self._anon_label(name)
- @util.memoized_property
- def _anon_key_label(self) -> _anonymous_label:
- """Provides a constant 'anonymous key label' for this ColumnElement.
- Compare to ``anon_label``, except that the "key" of the column,
- if available, is used to generate the label.
- This is used when a deduplicating key is placed into the columns
- collection of a selectable.
- .. versionchanged:: 1.4.9 - this attribute was not intended to be
- public and is renamed to _anon_key_label. anon_key_label exists
- for backwards compat
- """
- return self._anon_label(self._proxy_key)
- @property
- @util.deprecated(
- "1.4",
- "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
- "private, and the public accessor is deprecated.",
- )
- def anon_label(self) -> str:
- return self._anon_name_label
- @property
- @util.deprecated(
- "1.4",
- "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
- "now private, and the public accessor is deprecated.",
- )
- def anon_key_label(self) -> str:
- return self._anon_key_label
- def _dedupe_anon_label_idx(self, idx: int) -> str:
- """label to apply to a column that is anon labeled, but repeated
- in the SELECT, so that we have to make an "extra anon" label that
- disambiguates it from the previous appearance.
- these labels come out like "foo_bar_id__1" and have double underscores
- in them.
- """
- label = getattr(self, "name", None)
- # current convention is that if the element doesn't have a
- # ".name" (usually because it is not NamedColumn), we try to
- # use a "table qualified" form for the "dedupe anon" label,
- # based on the notion that a label like
- # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
- # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
- if label is None:
- return self._dedupe_anon_tq_label_idx(idx)
- else:
- return self._anon_label(label, add_hash=idx)
- @util.memoized_property
- def _anon_tq_label(self) -> _anonymous_label:
- return self._anon_label(getattr(self, "_tq_label", None))
- @util.memoized_property
- def _anon_tq_key_label(self) -> _anonymous_label:
- return self._anon_label(getattr(self, "_tq_key_label", None))
- def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
- label = getattr(self, "_tq_label", None) or "anon"
- return self._anon_label(label, add_hash=idx)
- class KeyedColumnElement(ColumnElement[_T]):
- """ColumnElement where ``.key`` is non-None."""
- _is_keyed_column_element = True
- key: str
- class WrapsColumnExpression(ColumnElement[_T]):
- """Mixin that defines a :class:`_expression.ColumnElement`
- as a wrapper with special
- labeling behavior for an expression that already has a name.
- .. versionadded:: 1.4
- .. seealso::
- :ref:`change_4449`
- """
- @property
- def wrapped_column_expression(self) -> ColumnElement[_T]:
- raise NotImplementedError()
- @util.non_memoized_property
- def _tq_label(self) -> Optional[str]:
- wce = self.wrapped_column_expression
- if hasattr(wce, "_tq_label"):
- return wce._tq_label
- else:
- return None
- @property
- def _label(self) -> Optional[str]:
- return self._tq_label
- @property
- def _non_anon_label(self) -> Optional[str]:
- return None
- @util.non_memoized_property
- def _anon_name_label(self) -> str:
- wce = self.wrapped_column_expression
- # this logic tries to get the WrappedColumnExpression to render
- # with "<expr> AS <name>", where "<name>" is the natural name
- # within the expression itself. e.g. "CAST(table.foo) AS foo".
- if not wce._is_text_clause:
- nal = wce._non_anon_label
- if nal:
- return nal
- elif hasattr(wce, "_anon_name_label"):
- return wce._anon_name_label
- return super()._anon_name_label
- def _dedupe_anon_label_idx(self, idx: int) -> str:
- wce = self.wrapped_column_expression
- nal = wce._non_anon_label
- if nal:
- return self._anon_label(nal + "_")
- else:
- return self._dedupe_anon_tq_label_idx(idx)
- @property
- def _proxy_key(self):
- wce = self.wrapped_column_expression
- if not wce._is_text_clause:
- return wce._proxy_key
- return super()._proxy_key
- class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
- r"""Represent a "bound expression".
- :class:`.BindParameter` is invoked explicitly using the
- :func:`.bindparam` function, as in::
- from sqlalchemy import bindparam
- stmt = select(users_table).where(
- users_table.c.name == bindparam("username")
- )
- Detailed discussion of how :class:`.BindParameter` is used is
- at :func:`.bindparam`.
- .. seealso::
- :func:`.bindparam`
- """
- __visit_name__ = "bindparam"
- _traverse_internals: _TraverseInternalsType = [
- ("key", InternalTraversal.dp_anon_name),
- ("type", InternalTraversal.dp_type),
- ("callable", InternalTraversal.dp_plain_dict),
- ("value", InternalTraversal.dp_plain_obj),
- ("literal_execute", InternalTraversal.dp_boolean),
- ]
- key: str
- type: TypeEngine[_T]
- value: Optional[_T]
- _is_crud = False
- _is_bind_parameter = True
- _key_is_anon = False
- # bindparam implements its own _gen_cache_key() method however
- # we check subclasses for this flag, else no cache key is generated
- inherit_cache = True
- def __init__(
- self,
- key: Optional[str],
- value: Any = _NoArg.NO_ARG,
- type_: Optional[_TypeEngineArgument[_T]] = None,
- unique: bool = False,
- required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
- quote: Optional[bool] = None,
- callable_: Optional[Callable[[], Any]] = None,
- expanding: bool = False,
- isoutparam: bool = False,
- literal_execute: bool = False,
- _compared_to_operator: Optional[OperatorType] = None,
- _compared_to_type: Optional[TypeEngine[Any]] = None,
- _is_crud: bool = False,
- ):
- if required is _NoArg.NO_ARG:
- required = value is _NoArg.NO_ARG and callable_ is None
- if value is _NoArg.NO_ARG:
- value = None
- if quote is not None:
- key = quoted_name.construct(key, quote)
- if unique:
- self.key = _anonymous_label.safe_construct(
- id(self),
- (
- key
- if key is not None
- and not isinstance(key, _anonymous_label)
- else "param"
- ),
- sanitize_key=True,
- )
- self._key_is_anon = True
- elif key:
- self.key = key
- else:
- self.key = _anonymous_label.safe_construct(id(self), "param")
- self._key_is_anon = True
- # identifying key that won't change across
- # clones, used to identify the bind's logical
- # identity
- self._identifying_key = self.key
- # key that was passed in the first place, used to
- # generate new keys
- self._orig_key = key or "param"
- self.unique = unique
- self.value = value
- self.callable = callable_
- self.isoutparam = isoutparam
- self.required = required
- # indicate an "expanding" parameter; the compiler sets this
- # automatically in the compiler _render_in_expr_w_bindparam method
- # for an IN expression
- self.expanding = expanding
- # this is another hint to help w/ expanding and is typically
- # set in the compiler _render_in_expr_w_bindparam method for an
- # IN expression
- self.expand_op = None
- self.literal_execute = literal_execute
- if _is_crud:
- self._is_crud = True
- if type_ is None:
- if expanding:
- if value:
- check_value = value[0]
- else:
- check_value = type_api._NO_VALUE_IN_LIST
- else:
- check_value = value
- if _compared_to_type is not None:
- self.type = _compared_to_type.coerce_compared_value(
- _compared_to_operator, check_value
- )
- else:
- self.type = type_api._resolve_value_to_type(check_value)
- elif isinstance(type_, type):
- self.type = type_()
- elif is_tuple_type(type_):
- if value:
- if expanding:
- check_value = value[0]
- else:
- check_value = value
- cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
- type_._resolve_values_to_types(check_value)
- )
- else:
- cast("BindParameter[typing_Tuple[Any, ...]]", self).type = (
- type_
- )
- else:
- self.type = type_
- def _with_value(self, value, maintain_key=False, required=NO_ARG):
- """Return a copy of this :class:`.BindParameter` with the given value
- set.
- """
- cloned = self._clone(maintain_key=maintain_key)
- cloned.value = value
- cloned.callable = None
- cloned.required = required if required is not NO_ARG else self.required
- if cloned.type is type_api.NULLTYPE:
- cloned.type = type_api._resolve_value_to_type(value)
- return cloned
- @property
- def effective_value(self) -> Optional[_T]:
- """Return the value of this bound parameter,
- taking into account if the ``callable`` parameter
- was set.
- The ``callable`` value will be evaluated
- and returned if present, else ``value``.
- """
- if self.callable:
- # TODO: set up protocol for bind parameter callable
- return self.callable() # type: ignore
- else:
- return self.value
- def render_literal_execute(self) -> BindParameter[_T]:
- """Produce a copy of this bound parameter that will enable the
- :paramref:`_sql.BindParameter.literal_execute` flag.
- The :paramref:`_sql.BindParameter.literal_execute` flag will
- have the effect of the parameter rendered in the compiled SQL
- string using ``[POSTCOMPILE]`` form, which is a special form that
- is converted to be a rendering of the literal value of the parameter
- at SQL execution time. The rationale is to support caching
- of SQL statement strings that can embed per-statement literal values,
- such as LIMIT and OFFSET parameters, in the final SQL string that
- is passed to the DBAPI. Dialects in particular may want to use
- this method within custom compilation schemes.
- .. versionadded:: 1.4.5
- .. seealso::
- :ref:`engine_thirdparty_caching`
- """
- c = ClauseElement._clone(self)
- c.literal_execute = True
- return c
- def _negate_in_binary(self, negated_op, original_op):
- if self.expand_op is original_op:
- bind = self._clone()
- bind.expand_op = negated_op
- return bind
- else:
- return self
- def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
- c: Self = ClauseElement._clone(self)
- c.type = type_
- return c
- def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
- c = ClauseElement._clone(self, **kw)
- # ensure all the BindParameter objects stay in cloned set.
- # in #7823, we changed "clone" so that a clone only keeps a reference
- # to the "original" element, since for column correspondence, that's
- # all we need. However, for BindParam, _cloned_set is used by
- # the "cache key bind match" lookup, which means if any of those
- # interim BindParameter objects became part of a cache key in the
- # cache, we need it. So here, make sure all clones keep carrying
- # forward.
- c._cloned_set.update(self._cloned_set)
- if not maintain_key and self.unique:
- c.key = _anonymous_label.safe_construct(
- id(c), c._orig_key or "param", sanitize_key=True
- )
- return c
- def _gen_cache_key(self, anon_map, bindparams):
- _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
- if not _gen_cache_ok:
- if anon_map is not None:
- anon_map[NO_CACHE] = True
- return None
- id_, found = anon_map.get_anon(self)
- if found:
- return (id_, self.__class__)
- if bindparams is not None:
- bindparams.append(self)
- return (
- id_,
- self.__class__,
- self.type._static_cache_key,
- self.key % anon_map if self._key_is_anon else self.key,
- self.literal_execute,
- )
- def _convert_to_unique(self):
- if not self.unique:
- self.unique = True
- self.key = _anonymous_label.safe_construct(
- id(self), self._orig_key or "param", sanitize_key=True
- )
- def __getstate__(self):
- """execute a deferred value for serialization purposes."""
- d = self.__dict__.copy()
- v = self.value
- if self.callable:
- v = self.callable()
- d["callable"] = None
- d["value"] = v
- return d
- def __setstate__(self, state):
- if state.get("unique", False):
- state["key"] = _anonymous_label.safe_construct(
- id(self), state.get("_orig_key", "param"), sanitize_key=True
- )
- self.__dict__.update(state)
- def __repr__(self):
- return "%s(%r, %r, type_=%r)" % (
- self.__class__.__name__,
- self.key,
- self.value,
- self.type,
- )
- class TypeClause(DQLDMLClauseElement):
- """Handle a type keyword in a SQL statement.
- Used by the ``Case`` statement.
- """
- __visit_name__ = "typeclause"
- _traverse_internals: _TraverseInternalsType = [
- ("type", InternalTraversal.dp_type)
- ]
- type: TypeEngine[Any]
- def __init__(self, type_: TypeEngine[Any]):
- self.type = type_
- class TextClause(
- roles.DDLConstraintColumnRole,
- roles.DDLExpressionRole,
- roles.StatementOptionRole,
- roles.WhereHavingRole,
- roles.OrderByRole,
- roles.FromClauseRole,
- roles.SelectStatementRole,
- roles.InElementRole,
- Generative,
- Executable,
- DQLDMLClauseElement,
- roles.BinaryElementRole[Any],
- inspection.Inspectable["TextClause"],
- ):
- """Represent a literal SQL text fragment.
- E.g.::
- from sqlalchemy import text
- t = text("SELECT * FROM users")
- result = connection.execute(t)
- The :class:`_expression.TextClause` construct is produced using the
- :func:`_expression.text`
- function; see that function for full documentation.
- .. seealso::
- :func:`_expression.text`
- """
- __visit_name__ = "textclause"
- _traverse_internals: _TraverseInternalsType = [
- ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
- ("text", InternalTraversal.dp_string),
- ]
- _is_text_clause = True
- _is_textual = True
- _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
- _is_implicitly_boolean = False
- _render_label_in_columns_clause = False
- _omit_from_statements = False
- _is_collection_aggregate = False
- @property
- def _hide_froms(self) -> Iterable[FromClause]:
- return ()
- def __and__(self, other):
- # support use in select.where(), query.filter()
- return and_(self, other)
- @property
- def _select_iterable(self) -> _SelectIterable:
- return (self,)
- # help in those cases where text() is
- # interpreted in a column expression situation
- key: Optional[str] = None
- _label: Optional[str] = None
- _allow_label_resolve = False
- @property
- def _is_star(self): # type: ignore[override]
- return self.text == "*"
- def __init__(self, text: str):
- self._bindparams: Dict[str, BindParameter[Any]] = {}
- def repl(m):
- self._bindparams[m.group(1)] = BindParameter(m.group(1))
- return ":%s" % m.group(1)
- # scan the string and search for bind parameter names, add them
- # to the list of bindparams
- self.text = self._bind_params_regex.sub(repl, text)
- @_generative
- def bindparams(
- self,
- *binds: BindParameter[Any],
- **names_to_values: Any,
- ) -> Self:
- """Establish the values and/or types of bound parameters within
- this :class:`_expression.TextClause` construct.
- Given a text construct such as::
- from sqlalchemy import text
- stmt = text(
- "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
- )
- the :meth:`_expression.TextClause.bindparams`
- method can be used to establish
- the initial value of ``:name`` and ``:timestamp``,
- using simple keyword arguments::
- stmt = stmt.bindparams(
- name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
- )
- Where above, new :class:`.BindParameter` objects
- will be generated with the names ``name`` and ``timestamp``, and
- values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
- respectively. The types will be
- inferred from the values given, in this case :class:`.String` and
- :class:`.DateTime`.
- When specific typing behavior is needed, the positional ``*binds``
- argument can be used in which to specify :func:`.bindparam` constructs
- directly. These constructs must include at least the ``key``
- argument, then an optional value and type::
- from sqlalchemy import bindparam
- stmt = stmt.bindparams(
- bindparam("name", value="jack", type_=String),
- bindparam("timestamp", type_=DateTime),
- )
- Above, we specified the type of :class:`.DateTime` for the
- ``timestamp`` bind, and the type of :class:`.String` for the ``name``
- bind. In the case of ``name`` we also set the default value of
- ``"jack"``.
- Additional bound parameters can be supplied at statement execution
- time, e.g.::
- result = connection.execute(
- stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
- )
- The :meth:`_expression.TextClause.bindparams`
- method can be called repeatedly,
- where it will re-use existing :class:`.BindParameter` objects to add
- new information. For example, we can call
- :meth:`_expression.TextClause.bindparams`
- first with typing information, and a
- second time with value information, and it will be combined::
- stmt = text(
- "SELECT id, name FROM user WHERE name=:name "
- "AND timestamp=:timestamp"
- )
- stmt = stmt.bindparams(
- bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
- )
- stmt = stmt.bindparams(
- name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
- )
- The :meth:`_expression.TextClause.bindparams`
- method also supports the concept of
- **unique** bound parameters. These are parameters that are
- "uniquified" on name at statement compilation time, so that multiple
- :func:`_expression.text`
- constructs may be combined together without the names
- conflicting. To use this feature, specify the
- :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
- object::
- stmt1 = text("select id from table where name=:name").bindparams(
- bindparam("name", value="name1", unique=True)
- )
- stmt2 = text("select id from table where name=:name").bindparams(
- bindparam("name", value="name2", unique=True)
- )
- union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
- The above statement will render as:
- .. sourcecode:: sql
- select id from table where name=:name_1
- UNION ALL select id from table where name=:name_2
- .. versionadded:: 1.3.11 Added support for the
- :paramref:`.BindParameter.unique` flag to work with
- :func:`_expression.text`
- constructs.
- """ # noqa: E501
- self._bindparams = new_params = self._bindparams.copy()
- for bind in binds:
- try:
- # the regex used for text() currently will not match
- # a unique/anonymous key in any case, so use the _orig_key
- # so that a text() construct can support unique parameters
- existing = new_params[bind._orig_key]
- except KeyError as err:
- raise exc.ArgumentError(
- "This text() construct doesn't define a "
- "bound parameter named %r" % bind._orig_key
- ) from err
- else:
- new_params[existing._orig_key] = bind
- for key, value in names_to_values.items():
- try:
- existing = new_params[key]
- except KeyError as err:
- raise exc.ArgumentError(
- "This text() construct doesn't define a "
- "bound parameter named %r" % key
- ) from err
- else:
- new_params[key] = existing._with_value(value, required=False)
- return self
- @util.preload_module("sqlalchemy.sql.selectable")
- def columns(
- self,
- *cols: _ColumnExpressionArgument[Any],
- **types: _TypeEngineArgument[Any],
- ) -> TextualSelect:
- r"""Turn this :class:`_expression.TextClause` object into a
- :class:`_expression.TextualSelect`
- object that serves the same role as a SELECT
- statement.
- The :class:`_expression.TextualSelect` is part of the
- :class:`_expression.SelectBase`
- hierarchy and can be embedded into another statement by using the
- :meth:`_expression.TextualSelect.subquery` method to produce a
- :class:`.Subquery`
- object, which can then be SELECTed from.
- This function essentially bridges the gap between an entirely
- textual SELECT statement and the SQL expression language concept
- of a "selectable"::
- from sqlalchemy.sql import column, text
- stmt = text("SELECT id, name FROM some_table")
- stmt = stmt.columns(column("id"), column("name")).subquery("st")
- stmt = (
- select(mytable)
- .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
- .where(stmt.c.id > 5)
- )
- Above, we pass a series of :func:`_expression.column` elements to the
- :meth:`_expression.TextClause.columns` method positionally. These
- :func:`_expression.column`
- elements now become first class elements upon the
- :attr:`_expression.TextualSelect.selected_columns` column collection,
- which then
- become part of the :attr:`.Subquery.c` collection after
- :meth:`_expression.TextualSelect.subquery` is invoked.
- The column expressions we pass to
- :meth:`_expression.TextClause.columns` may
- also be typed; when we do so, these :class:`.TypeEngine` objects become
- the effective return type of the column, so that SQLAlchemy's
- result-set-processing systems may be used on the return values.
- This is often needed for types such as date or boolean types, as well
- as for unicode processing on some dialect configurations::
- stmt = text("SELECT id, name, timestamp FROM some_table")
- stmt = stmt.columns(
- column("id", Integer),
- column("name", Unicode),
- column("timestamp", DateTime),
- )
- for id, name, timestamp in connection.execute(stmt):
- print(id, name, timestamp)
- As a shortcut to the above syntax, keyword arguments referring to
- types alone may be used, if only type conversion is needed::
- stmt = text("SELECT id, name, timestamp FROM some_table")
- stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
- for id, name, timestamp in connection.execute(stmt):
- print(id, name, timestamp)
- The positional form of :meth:`_expression.TextClause.columns`
- also provides the
- unique feature of **positional column targeting**, which is
- particularly useful when using the ORM with complex textual queries. If
- we specify the columns from our model to
- :meth:`_expression.TextClause.columns`,
- the result set will match to those columns positionally, meaning the
- name or origin of the column in the textual SQL doesn't matter::
- stmt = text(
- "SELECT users.id, addresses.id, users.id, "
- "users.name, addresses.email_address AS email "
- "FROM users JOIN addresses ON users.id=addresses.user_id "
- "WHERE users.id = 1"
- ).columns(
- User.id,
- Address.id,
- Address.user_id,
- User.name,
- Address.email_address,
- )
- query = (
- session.query(User)
- .from_statement(stmt)
- .options(contains_eager(User.addresses))
- )
- The :meth:`_expression.TextClause.columns` method provides a direct
- route to calling :meth:`_expression.FromClause.subquery` as well as
- :meth:`_expression.SelectBase.cte`
- against a textual SELECT statement::
- stmt = stmt.columns(id=Integer, name=String).cte("st")
- stmt = select(sometable).where(sometable.c.id == stmt.c.id)
- :param \*cols: A series of :class:`_expression.ColumnElement` objects,
- typically
- :class:`_schema.Column` objects from a :class:`_schema.Table`
- or ORM level
- column-mapped attributes, representing a set of columns that this
- textual string will SELECT from.
- :param \**types: A mapping of string names to :class:`.TypeEngine`
- type objects indicating the datatypes to use for names that are
- SELECTed from the textual string. Prefer to use the ``*cols``
- argument as it also indicates positional ordering.
- """
- selectable = util.preloaded.sql_selectable
- input_cols: List[NamedColumn[Any]] = [
- coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
- ]
- positional_input_cols = [
- (
- ColumnClause(col.key, types.pop(col.key))
- if col.key in types
- else col
- )
- for col in input_cols
- ]
- keyed_input_cols: List[NamedColumn[Any]] = [
- ColumnClause(key, type_) for key, type_ in types.items()
- ]
- elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
- elem._init(
- self,
- positional_input_cols + keyed_input_cols,
- positional=bool(positional_input_cols) and not keyed_input_cols,
- )
- return elem
- @property
- def type(self) -> TypeEngine[Any]:
- return type_api.NULLTYPE
- @property
- def comparator(self):
- # TODO: this seems wrong, it seems like we might not
- # be using this method.
- return self.type.comparator_factory(self) # type: ignore
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> Union[Self, Grouping[Any]]:
- if against is operators.in_op:
- return Grouping(self)
- else:
- return self
- class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
- """Represent the NULL keyword in a SQL statement.
- :class:`.Null` is accessed as a constant via the
- :func:`.null` function.
- """
- __visit_name__ = "null"
- _traverse_internals: _TraverseInternalsType = []
- _singleton: Null
- if not TYPE_CHECKING:
- @util.memoized_property
- def type(self) -> TypeEngine[_T]: # noqa: A001
- return type_api.NULLTYPE
- @classmethod
- def _instance(cls) -> Null:
- """Return a constant :class:`.Null` construct."""
- return Null._singleton
- Null._create_singleton()
- class False_(
- SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
- ):
- """Represent the ``false`` keyword, or equivalent, in a SQL statement.
- :class:`.False_` is accessed as a constant via the
- :func:`.false` function.
- """
- __visit_name__ = "false"
- _traverse_internals: _TraverseInternalsType = []
- _singleton: False_
- if not TYPE_CHECKING:
- @util.memoized_property
- def type(self) -> TypeEngine[_T]: # noqa: A001
- return type_api.BOOLEANTYPE
- def _negate(self) -> True_:
- return True_._singleton
- @classmethod
- def _instance(cls) -> False_:
- return False_._singleton
- False_._create_singleton()
- class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
- """Represent the ``true`` keyword, or equivalent, in a SQL statement.
- :class:`.True_` is accessed as a constant via the
- :func:`.true` function.
- """
- __visit_name__ = "true"
- _traverse_internals: _TraverseInternalsType = []
- _singleton: True_
- if not TYPE_CHECKING:
- @util.memoized_property
- def type(self) -> TypeEngine[_T]: # noqa: A001
- return type_api.BOOLEANTYPE
- def _negate(self) -> False_:
- return False_._singleton
- @classmethod
- def _ifnone(
- cls, other: Optional[ColumnElement[Any]]
- ) -> ColumnElement[Any]:
- if other is None:
- return cls._instance()
- else:
- return other
- @classmethod
- def _instance(cls) -> True_:
- return True_._singleton
- True_._create_singleton()
- class ClauseList(
- roles.InElementRole,
- roles.OrderByRole,
- roles.ColumnsClauseRole,
- roles.DMLColumnRole,
- DQLDMLClauseElement,
- ):
- """Describe a list of clauses, separated by an operator.
- By default, is comma-separated, such as a column listing.
- """
- __visit_name__ = "clauselist"
- # this is used only by the ORM in a legacy use case for
- # composite attributes
- _is_clause_list = True
- _traverse_internals: _TraverseInternalsType = [
- ("clauses", InternalTraversal.dp_clauseelement_list),
- ("operator", InternalTraversal.dp_operator),
- ]
- clauses: List[ColumnElement[Any]]
- def __init__(
- self,
- *clauses: _ColumnExpressionArgument[Any],
- operator: OperatorType = operators.comma_op,
- group: bool = True,
- group_contents: bool = True,
- _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
- ):
- self.operator = operator
- self.group = group
- self.group_contents = group_contents
- clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
- text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
- self._text_converter_role = text_converter_role
- if self.group_contents:
- self.clauses = [
- coercions.expect(
- text_converter_role, clause, apply_propagate_attrs=self
- ).self_group(against=self.operator)
- for clause in clauses_iterator
- ]
- else:
- self.clauses = [
- coercions.expect(
- text_converter_role, clause, apply_propagate_attrs=self
- )
- for clause in clauses_iterator
- ]
- self._is_implicitly_boolean = operators.is_boolean(self.operator)
- @classmethod
- def _construct_raw(
- cls,
- operator: OperatorType,
- clauses: Optional[Sequence[ColumnElement[Any]]] = None,
- ) -> ClauseList:
- self = cls.__new__(cls)
- self.clauses = list(clauses) if clauses else []
- self.group = True
- self.operator = operator
- self.group_contents = True
- self._is_implicitly_boolean = False
- return self
- def __iter__(self) -> Iterator[ColumnElement[Any]]:
- return iter(self.clauses)
- def __len__(self) -> int:
- return len(self.clauses)
- @property
- def _select_iterable(self) -> _SelectIterable:
- return itertools.chain.from_iterable(
- [elem._select_iterable for elem in self.clauses]
- )
- def append(self, clause):
- if self.group_contents:
- self.clauses.append(
- coercions.expect(self._text_converter_role, clause).self_group(
- against=self.operator
- )
- )
- else:
- self.clauses.append(
- coercions.expect(self._text_converter_role, clause)
- )
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return list(itertools.chain(*[c._from_objects for c in self.clauses]))
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> Union[Self, Grouping[Any]]:
- if self.group and operators.is_precedent(self.operator, against):
- return Grouping(self)
- else:
- return self
- class OperatorExpression(ColumnElement[_T]):
- """base for expressions that contain an operator and operands
- .. versionadded:: 2.0
- """
- operator: OperatorType
- type: TypeEngine[_T]
- group: bool = True
- @property
- def is_comparison(self):
- return operators.is_comparison(self.operator)
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> Union[Self, Grouping[_T]]:
- if (
- self.group
- and operators.is_precedent(self.operator, against)
- or (
- # a negate against a non-boolean operator
- # doesn't make too much sense but we should
- # group for that
- against is operators.inv
- and not operators.is_boolean(self.operator)
- )
- ):
- return Grouping(self)
- else:
- return self
- @property
- def _flattened_operator_clauses(
- self,
- ) -> typing_Tuple[ColumnElement[Any], ...]:
- raise NotImplementedError()
- @classmethod
- def _construct_for_op(
- cls,
- left: ColumnElement[Any],
- right: ColumnElement[Any],
- op: OperatorType,
- *,
- type_: TypeEngine[_T],
- negate: Optional[OperatorType] = None,
- modifiers: Optional[Mapping[str, Any]] = None,
- ) -> OperatorExpression[_T]:
- if operators.is_associative(op):
- assert (
- negate is None
- ), f"negate not supported for associative operator {op}"
- multi = False
- if getattr(
- left, "operator", None
- ) is op and type_._compare_type_affinity(left.type):
- multi = True
- left_flattened = left._flattened_operator_clauses
- else:
- left_flattened = (left,)
- if getattr(
- right, "operator", None
- ) is op and type_._compare_type_affinity(right.type):
- multi = True
- right_flattened = right._flattened_operator_clauses
- else:
- right_flattened = (right,)
- if multi:
- return ExpressionClauseList._construct_for_list(
- op,
- type_,
- *(left_flattened + right_flattened),
- )
- if right._is_collection_aggregate:
- negate = None
- return BinaryExpression(
- left, right, op, type_=type_, negate=negate, modifiers=modifiers
- )
- class ExpressionClauseList(OperatorExpression[_T]):
- """Describe a list of clauses, separated by an operator,
- in a column expression context.
- :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
- it represents a column-oriented DQL expression only, not an open ended
- list of anything comma separated.
- .. versionadded:: 2.0
- """
- __visit_name__ = "expression_clauselist"
- _traverse_internals: _TraverseInternalsType = [
- ("clauses", InternalTraversal.dp_clauseelement_tuple),
- ("operator", InternalTraversal.dp_operator),
- ]
- clauses: typing_Tuple[ColumnElement[Any], ...]
- group: bool
- def __init__(
- self,
- operator: OperatorType,
- *clauses: _ColumnExpressionArgument[Any],
- type_: Optional[_TypeEngineArgument[_T]] = None,
- ):
- self.operator = operator
- self.clauses = tuple(
- coercions.expect(
- roles.ExpressionElementRole, clause, apply_propagate_attrs=self
- )
- for clause in clauses
- )
- self._is_implicitly_boolean = operators.is_boolean(self.operator)
- self.type = type_api.to_instance(type_) # type: ignore
- @property
- def _flattened_operator_clauses(
- self,
- ) -> typing_Tuple[ColumnElement[Any], ...]:
- return self.clauses
- def __iter__(self) -> Iterator[ColumnElement[Any]]:
- return iter(self.clauses)
- def __len__(self) -> int:
- return len(self.clauses)
- @property
- def _select_iterable(self) -> _SelectIterable:
- return (self,)
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return list(itertools.chain(*[c._from_objects for c in self.clauses]))
- def _append_inplace(self, clause: ColumnElement[Any]) -> None:
- self.clauses += (clause,)
- @classmethod
- def _construct_for_list(
- cls,
- operator: OperatorType,
- type_: TypeEngine[_T],
- *clauses: ColumnElement[Any],
- group: bool = True,
- ) -> ExpressionClauseList[_T]:
- self = cls.__new__(cls)
- self.group = group
- if group:
- self.clauses = tuple(
- c.self_group(against=operator) for c in clauses
- )
- else:
- self.clauses = clauses
- self.operator = operator
- self.type = type_
- for c in clauses:
- if c._propagate_attrs:
- self._propagate_attrs = c._propagate_attrs
- break
- return self
- def _negate(self) -> Any:
- grouped = self.self_group(against=operators.inv)
- assert isinstance(grouped, ColumnElement)
- return UnaryExpression(grouped, operator=operators.inv)
- class BooleanClauseList(ExpressionClauseList[bool]):
- __visit_name__ = "expression_clauselist"
- inherit_cache = True
- def __init__(self, *arg, **kw):
- raise NotImplementedError(
- "BooleanClauseList has a private constructor"
- )
- @classmethod
- def _process_clauses_for_boolean(
- cls,
- operator: OperatorType,
- continue_on: Any,
- skip_on: Any,
- clauses: Iterable[ColumnElement[Any]],
- ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
- has_continue_on = None
- convert_clauses = []
- against = operators._asbool
- lcc = 0
- for clause in clauses:
- if clause is continue_on:
- # instance of continue_on, like and_(x, y, True, z), store it
- # if we didn't find one already, we will use it if there
- # are no other expressions here.
- has_continue_on = clause
- elif clause is skip_on:
- # instance of skip_on, e.g. and_(x, y, False, z), cancels
- # the rest out
- convert_clauses = [clause]
- lcc = 1
- break
- else:
- if not lcc:
- lcc = 1
- else:
- against = operator
- # technically this would be len(convert_clauses) + 1
- # however this only needs to indicate "greater than one"
- lcc = 2
- convert_clauses.append(clause)
- if not convert_clauses and has_continue_on is not None:
- convert_clauses = [has_continue_on]
- lcc = 1
- return lcc, [c.self_group(against=against) for c in convert_clauses]
- @classmethod
- def _construct(
- cls,
- operator: OperatorType,
- continue_on: Any,
- skip_on: Any,
- initial_clause: Any = _NoArg.NO_ARG,
- *clauses: Any,
- **kw: Any,
- ) -> ColumnElement[Any]:
- if initial_clause is _NoArg.NO_ARG:
- # no elements period. deprecated use case. return an empty
- # ClauseList construct that generates nothing unless it has
- # elements added to it.
- name = operator.__name__
- util.warn_deprecated(
- f"Invoking {name}() without arguments is deprecated, and "
- f"will be disallowed in a future release. For an empty "
- f"""{name}() construct, use '{name}({
- 'true()' if continue_on is True_._singleton else 'false()'
- }, *args)' """
- f"""or '{name}({
- 'True' if continue_on is True_._singleton else 'False'
- }, *args)'.""",
- version="1.4",
- )
- return cls._construct_raw(operator)
- lcc, convert_clauses = cls._process_clauses_for_boolean(
- operator,
- continue_on,
- skip_on,
- [
- coercions.expect(roles.WhereHavingRole, clause)
- for clause in util.coerce_generator_arg(
- (initial_clause,) + clauses
- )
- ],
- )
- if lcc > 1:
- # multiple elements. Return regular BooleanClauseList
- # which will link elements against the operator.
- flattened_clauses = itertools.chain.from_iterable(
- (
- (c for c in to_flat._flattened_operator_clauses)
- if getattr(to_flat, "operator", None) is operator
- else (to_flat,)
- )
- for to_flat in convert_clauses
- )
- return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
- else:
- assert lcc
- # just one element. return it as a single boolean element,
- # not a list and discard the operator.
- return convert_clauses[0]
- @classmethod
- def _construct_for_whereclause(
- cls, clauses: Iterable[ColumnElement[Any]]
- ) -> Optional[ColumnElement[bool]]:
- operator, continue_on, skip_on = (
- operators.and_,
- True_._singleton,
- False_._singleton,
- )
- lcc, convert_clauses = cls._process_clauses_for_boolean(
- operator,
- continue_on,
- skip_on,
- clauses, # these are assumed to be coerced already
- )
- if lcc > 1:
- # multiple elements. Return regular BooleanClauseList
- # which will link elements against the operator.
- return cls._construct_raw(operator, convert_clauses)
- elif lcc == 1:
- # just one element. return it as a single boolean element,
- # not a list and discard the operator.
- return convert_clauses[0]
- else:
- return None
- @classmethod
- def _construct_raw(
- cls,
- operator: OperatorType,
- clauses: Optional[Sequence[ColumnElement[Any]]] = None,
- ) -> BooleanClauseList:
- self = cls.__new__(cls)
- self.clauses = tuple(clauses) if clauses else ()
- self.group = True
- self.operator = operator
- self.type = type_api.BOOLEANTYPE
- self._is_implicitly_boolean = True
- return self
- @classmethod
- def and_(
- cls,
- initial_clause: Union[
- Literal[True], _ColumnExpressionArgument[bool], _NoArg
- ] = _NoArg.NO_ARG,
- *clauses: _ColumnExpressionArgument[bool],
- ) -> ColumnElement[bool]:
- r"""Produce a conjunction of expressions joined by ``AND``.
- See :func:`_sql.and_` for full documentation.
- """
- return cls._construct(
- operators.and_,
- True_._singleton,
- False_._singleton,
- initial_clause,
- *clauses,
- )
- @classmethod
- def or_(
- cls,
- initial_clause: Union[
- Literal[False], _ColumnExpressionArgument[bool], _NoArg
- ] = _NoArg.NO_ARG,
- *clauses: _ColumnExpressionArgument[bool],
- ) -> ColumnElement[bool]:
- """Produce a conjunction of expressions joined by ``OR``.
- See :func:`_sql.or_` for full documentation.
- """
- return cls._construct(
- operators.or_,
- False_._singleton,
- True_._singleton,
- initial_clause,
- *clauses,
- )
- @property
- def _select_iterable(self) -> _SelectIterable:
- return (self,)
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> Union[Self, Grouping[bool]]:
- if not self.clauses:
- return self
- else:
- return super().self_group(against=against)
- and_ = BooleanClauseList.and_
- or_ = BooleanClauseList.or_
- class Tuple(ClauseList, ColumnElement[typing_Tuple[Any, ...]]):
- """Represent a SQL tuple."""
- __visit_name__ = "tuple"
- _traverse_internals: _TraverseInternalsType = (
- ClauseList._traverse_internals + []
- )
- type: TupleType
- @util.preload_module("sqlalchemy.sql.sqltypes")
- def __init__(
- self,
- *clauses: _ColumnExpressionArgument[Any],
- types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
- ):
- sqltypes = util.preloaded.sql_sqltypes
- if types is None:
- init_clauses: List[ColumnElement[Any]] = [
- coercions.expect(roles.ExpressionElementRole, c)
- for c in clauses
- ]
- else:
- if len(types) != len(clauses):
- raise exc.ArgumentError(
- "Wrong number of elements for %d-tuple: %r "
- % (len(types), clauses)
- )
- init_clauses = [
- coercions.expect(
- roles.ExpressionElementRole,
- c,
- type_=typ if not typ._isnull else None,
- )
- for typ, c in zip(types, clauses)
- ]
- self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
- super().__init__(*init_clauses)
- @property
- def _select_iterable(self) -> _SelectIterable:
- return (self,)
- def _bind_param(self, operator, obj, type_=None, expanding=False):
- if expanding:
- return BindParameter(
- None,
- value=obj,
- _compared_to_operator=operator,
- unique=True,
- expanding=True,
- type_=type_,
- _compared_to_type=self.type,
- )
- else:
- return Tuple(
- *[
- BindParameter(
- None,
- o,
- _compared_to_operator=operator,
- _compared_to_type=compared_to_type,
- unique=True,
- type_=type_,
- )
- for o, compared_to_type in zip(obj, self.type.types)
- ]
- )
- def self_group(self, against: Optional[OperatorType] = None) -> Self:
- # Tuple is parenthesized by definition.
- return self
- class Case(ColumnElement[_T]):
- """Represent a ``CASE`` expression.
- :class:`.Case` is produced using the :func:`.case` factory function,
- as in::
- from sqlalchemy import case
- stmt = select(users_table).where(
- case(
- (users_table.c.name == "wendy", "W"),
- (users_table.c.name == "jack", "J"),
- else_="E",
- )
- )
- Details on :class:`.Case` usage is at :func:`.case`.
- .. seealso::
- :func:`.case`
- """
- __visit_name__ = "case"
- _traverse_internals: _TraverseInternalsType = [
- ("value", InternalTraversal.dp_clauseelement),
- ("whens", InternalTraversal.dp_clauseelement_tuples),
- ("else_", InternalTraversal.dp_clauseelement),
- ]
- # for case(), the type is derived from the whens. so for the moment
- # users would have to cast() the case to get a specific type
- whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
- else_: Optional[ColumnElement[_T]]
- value: Optional[ColumnElement[Any]]
- def __init__(
- self,
- *whens: Union[
- typing_Tuple[_ColumnExpressionArgument[bool], Any],
- Mapping[Any, Any],
- ],
- value: Optional[Any] = None,
- else_: Optional[Any] = None,
- ):
- new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
- "whens", "case", whens
- )
- try:
- new_whens = util.dictlike_iteritems(new_whens)
- except TypeError:
- pass
- self.whens = [
- (
- coercions.expect(
- roles.ExpressionElementRole,
- c,
- apply_propagate_attrs=self,
- ).self_group(),
- coercions.expect(roles.ExpressionElementRole, r),
- )
- for (c, r) in new_whens
- ]
- if value is None:
- self.value = None
- else:
- self.value = coercions.expect(roles.ExpressionElementRole, value)
- if else_ is not None:
- self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
- else:
- self.else_ = None
- type_ = next(
- (
- then.type
- # Iterate `whens` in reverse to match previous behaviour
- # where type of final element took priority
- for *_, then in reversed(self.whens)
- if not then.type._isnull
- ),
- self.else_.type if self.else_ is not None else type_api.NULLTYPE,
- )
- self.type = cast(_T, type_)
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return list(
- itertools.chain(*[x._from_objects for x in self.get_children()])
- )
- class Cast(WrapsColumnExpression[_T]):
- """Represent a ``CAST`` expression.
- :class:`.Cast` is produced using the :func:`.cast` factory function,
- as in::
- from sqlalchemy import cast, Numeric
- stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
- Details on :class:`.Cast` usage is at :func:`.cast`.
- .. seealso::
- :ref:`tutorial_casts`
- :func:`.cast`
- :func:`.try_cast`
- :func:`.type_coerce` - an alternative to CAST that coerces the type
- on the Python side only, which is often sufficient to generate the
- correct SQL and data coercion.
- """
- __visit_name__ = "cast"
- _traverse_internals: _TraverseInternalsType = [
- ("clause", InternalTraversal.dp_clauseelement),
- ("type", InternalTraversal.dp_type),
- ]
- clause: ColumnElement[Any]
- type: TypeEngine[_T]
- typeclause: TypeClause
- def __init__(
- self,
- expression: _ColumnExpressionArgument[Any],
- type_: _TypeEngineArgument[_T],
- ):
- self.type = type_api.to_instance(type_)
- self.clause = coercions.expect(
- roles.ExpressionElementRole,
- expression,
- type_=self.type,
- apply_propagate_attrs=self,
- )
- self.typeclause = TypeClause(self.type)
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return self.clause._from_objects
- @property
- def wrapped_column_expression(self):
- return self.clause
- class TryCast(Cast[_T]):
- """Represent a TRY_CAST expression.
- Details on :class:`.TryCast` usage is at :func:`.try_cast`.
- .. seealso::
- :func:`.try_cast`
- :ref:`tutorial_casts`
- """
- __visit_name__ = "try_cast"
- inherit_cache = True
- class TypeCoerce(WrapsColumnExpression[_T]):
- """Represent a Python-side type-coercion wrapper.
- :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
- function; see that function for usage details.
- .. seealso::
- :func:`_expression.type_coerce`
- :func:`.cast`
- """
- __visit_name__ = "type_coerce"
- _traverse_internals: _TraverseInternalsType = [
- ("clause", InternalTraversal.dp_clauseelement),
- ("type", InternalTraversal.dp_type),
- ]
- clause: ColumnElement[Any]
- type: TypeEngine[_T]
- def __init__(
- self,
- expression: _ColumnExpressionArgument[Any],
- type_: _TypeEngineArgument[_T],
- ):
- self.type = type_api.to_instance(type_)
- self.clause = coercions.expect(
- roles.ExpressionElementRole,
- expression,
- type_=self.type,
- apply_propagate_attrs=self,
- )
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return self.clause._from_objects
- @HasMemoized.memoized_attribute
- def typed_expression(self):
- if isinstance(self.clause, BindParameter):
- bp = self.clause._clone()
- bp.type = self.type
- return bp
- else:
- return self.clause
- @property
- def wrapped_column_expression(self):
- return self.clause
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> TypeCoerce[_T]:
- grouped = self.clause.self_group(against=against)
- if grouped is not self.clause:
- return TypeCoerce(grouped, self.type)
- else:
- return self
- class Extract(ColumnElement[int]):
- """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
- __visit_name__ = "extract"
- _traverse_internals: _TraverseInternalsType = [
- ("expr", InternalTraversal.dp_clauseelement),
- ("field", InternalTraversal.dp_string),
- ]
- expr: ColumnElement[Any]
- field: str
- def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
- self.type = type_api.INTEGERTYPE
- self.field = field
- self.expr = coercions.expect(roles.ExpressionElementRole, expr)
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return self.expr._from_objects
- class _label_reference(ColumnElement[_T]):
- """Wrap a column expression as it appears in a 'reference' context.
- This expression is any that includes an _order_by_label_element,
- which is a Label, or a DESC / ASC construct wrapping a Label.
- The production of _label_reference() should occur when an expression
- is added to this context; this includes the ORDER BY or GROUP BY of a
- SELECT statement, as well as a few other places, such as the ORDER BY
- within an OVER clause.
- """
- __visit_name__ = "label_reference"
- _traverse_internals: _TraverseInternalsType = [
- ("element", InternalTraversal.dp_clauseelement)
- ]
- element: ColumnElement[_T]
- def __init__(self, element: ColumnElement[_T]):
- self.element = element
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return []
- class _textual_label_reference(ColumnElement[Any]):
- __visit_name__ = "textual_label_reference"
- _traverse_internals: _TraverseInternalsType = [
- ("element", InternalTraversal.dp_string)
- ]
- def __init__(self, element: str):
- self.element = element
- @util.memoized_property
- def _text_clause(self) -> TextClause:
- return TextClause(self.element)
- class UnaryExpression(ColumnElement[_T]):
- """Define a 'unary' expression.
- A unary expression has a single column expression
- and an operator. The operator can be placed on the left
- (where it is called the 'operator') or right (where it is called the
- 'modifier') of the column expression.
- :class:`.UnaryExpression` is the basis for several unary operators
- including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
- :func:`.nulls_first` and :func:`.nulls_last`.
- """
- __visit_name__ = "unary"
- _traverse_internals: _TraverseInternalsType = [
- ("element", InternalTraversal.dp_clauseelement),
- ("operator", InternalTraversal.dp_operator),
- ("modifier", InternalTraversal.dp_operator),
- ]
- element: ColumnElement[Any]
- operator: Optional[OperatorType]
- modifier: Optional[OperatorType]
- def __init__(
- self,
- element: ColumnElement[Any],
- *,
- operator: Optional[OperatorType] = None,
- modifier: Optional[OperatorType] = None,
- type_: Optional[_TypeEngineArgument[_T]] = None,
- wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
- ):
- self.operator = operator
- self.modifier = modifier
- self._propagate_attrs = element._propagate_attrs
- self.element = element.self_group(
- against=self.operator or self.modifier
- )
- # if type is None, we get NULLTYPE, which is our _T. But I don't
- # know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
- def _wraps_unnamed_column(self):
- ungrouped = self.element._ungroup()
- return (
- not isinstance(ungrouped, NamedColumn)
- or ungrouped._non_anon_label is None
- )
- @classmethod
- def _create_nulls_first(
- cls,
- column: _ColumnExpressionArgument[_T],
- ) -> UnaryExpression[_T]:
- return UnaryExpression(
- coercions.expect(roles.ByOfRole, column),
- modifier=operators.nulls_first_op,
- )
- @classmethod
- def _create_nulls_last(
- cls,
- column: _ColumnExpressionArgument[_T],
- ) -> UnaryExpression[_T]:
- return UnaryExpression(
- coercions.expect(roles.ByOfRole, column),
- modifier=operators.nulls_last_op,
- )
- @classmethod
- def _create_desc(
- cls, column: _ColumnExpressionOrStrLabelArgument[_T]
- ) -> UnaryExpression[_T]:
- return UnaryExpression(
- coercions.expect(roles.ByOfRole, column),
- modifier=operators.desc_op,
- )
- @classmethod
- def _create_asc(
- cls,
- column: _ColumnExpressionOrStrLabelArgument[_T],
- ) -> UnaryExpression[_T]:
- return UnaryExpression(
- coercions.expect(roles.ByOfRole, column),
- modifier=operators.asc_op,
- )
- @classmethod
- def _create_distinct(
- cls,
- expr: _ColumnExpressionArgument[_T],
- ) -> UnaryExpression[_T]:
- col_expr: ColumnElement[_T] = coercions.expect(
- roles.ExpressionElementRole, expr
- )
- return UnaryExpression(
- col_expr,
- operator=operators.distinct_op,
- type_=col_expr.type,
- )
- @classmethod
- def _create_bitwise_not(
- cls,
- expr: _ColumnExpressionArgument[_T],
- ) -> UnaryExpression[_T]:
- col_expr: ColumnElement[_T] = coercions.expect(
- roles.ExpressionElementRole, expr
- )
- return UnaryExpression(
- col_expr,
- operator=operators.bitwise_not_op,
- type_=col_expr.type,
- )
- @property
- def _order_by_label_element(self) -> Optional[Label[Any]]:
- if operators.is_order_by_modifier(self.modifier):
- return self.element._order_by_label_element
- else:
- return None
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return self.element._from_objects
- def _negate(self) -> ColumnElement[Any]:
- if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
- return UnaryExpression(
- self.self_group(against=operators.inv),
- operator=operators.inv,
- type_=type_api.BOOLEANTYPE,
- )
- else:
- return ColumnElement._negate(self)
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> Union[Self, Grouping[_T]]:
- if self.operator and operators.is_precedent(self.operator, against):
- return Grouping(self)
- else:
- return self
- class CollectionAggregate(UnaryExpression[_T]):
- """Forms the basis for right-hand collection operator modifiers
- ANY and ALL.
- The ANY and ALL keywords are available in different ways on different
- backends. On PostgreSQL, they only work for an ARRAY type. On
- MySQL, they only work for subqueries.
- """
- inherit_cache = True
- _is_collection_aggregate = True
- @classmethod
- def _create_any(
- cls, expr: _ColumnExpressionArgument[_T]
- ) -> CollectionAggregate[bool]:
- col_expr: ColumnElement[_T] = coercions.expect(
- roles.ExpressionElementRole,
- expr,
- )
- col_expr = col_expr.self_group()
- return CollectionAggregate(
- col_expr,
- operator=operators.any_op,
- type_=type_api.BOOLEANTYPE,
- )
- @classmethod
- def _create_all(
- cls, expr: _ColumnExpressionArgument[_T]
- ) -> CollectionAggregate[bool]:
- col_expr: ColumnElement[_T] = coercions.expect(
- roles.ExpressionElementRole,
- expr,
- )
- col_expr = col_expr.self_group()
- return CollectionAggregate(
- col_expr,
- operator=operators.all_op,
- type_=type_api.BOOLEANTYPE,
- )
- # operate and reverse_operate are hardwired to
- # dispatch onto the type comparator directly, so that we can
- # ensure "reversed" behavior.
- def operate(
- self, op: OperatorType, *other: Any, **kwargs: Any
- ) -> ColumnElement[_T]:
- if not operators.is_comparison(op):
- raise exc.ArgumentError(
- "Only comparison operators may be used with ANY/ALL"
- )
- kwargs["reverse"] = True
- return self.comparator.operate(operators.mirror(op), *other, **kwargs)
- def reverse_operate(
- self, op: OperatorType, other: Any, **kwargs: Any
- ) -> ColumnElement[_T]:
- # comparison operators should never call reverse_operate
- assert not operators.is_comparison(op)
- raise exc.ArgumentError(
- "Only comparison operators may be used with ANY/ALL"
- )
- class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
- inherit_cache = True
- def __init__(self, element, operator, negate):
- self.element = element
- self.type = type_api.BOOLEANTYPE
- self.operator = operator
- self.negate = negate
- self.modifier = None
- self._is_implicitly_boolean = element._is_implicitly_boolean
- @property
- def wrapped_column_expression(self):
- return self.element
- def self_group(self, against: Optional[OperatorType] = None) -> Self:
- return self
- def _negate(self):
- if isinstance(self.element, (True_, False_)):
- return self.element._negate()
- else:
- return AsBoolean(self.element, self.negate, self.operator)
- class BinaryExpression(OperatorExpression[_T]):
- """Represent an expression that is ``LEFT <operator> RIGHT``.
- A :class:`.BinaryExpression` is generated automatically
- whenever two column expressions are used in a Python binary expression:
- .. sourcecode:: pycon+sql
- >>> from sqlalchemy.sql import column
- >>> column("a") + column("b")
- <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
- >>> print(column("a") + column("b"))
- {printsql}a + b
- """
- __visit_name__ = "binary"
- _traverse_internals: _TraverseInternalsType = [
- ("left", InternalTraversal.dp_clauseelement),
- ("right", InternalTraversal.dp_clauseelement),
- ("operator", InternalTraversal.dp_operator),
- ("negate", InternalTraversal.dp_operator),
- ("modifiers", InternalTraversal.dp_plain_dict),
- (
- "type",
- InternalTraversal.dp_type,
- ),
- ]
- _cache_key_traversal = [
- ("left", InternalTraversal.dp_clauseelement),
- ("right", InternalTraversal.dp_clauseelement),
- ("operator", InternalTraversal.dp_operator),
- ("modifiers", InternalTraversal.dp_plain_dict),
- # "type" affects JSON CAST operators, so while redundant in most cases,
- # is needed for that one
- (
- "type",
- InternalTraversal.dp_type,
- ),
- ]
- _is_implicitly_boolean = True
- """Indicates that any database will know this is a boolean expression
- even if the database does not have an explicit boolean datatype.
- """
- left: ColumnElement[Any]
- right: ColumnElement[Any]
- modifiers: Mapping[str, Any]
- def __init__(
- self,
- left: ColumnElement[Any],
- right: ColumnElement[Any],
- operator: OperatorType,
- type_: Optional[_TypeEngineArgument[_T]] = None,
- negate: Optional[OperatorType] = None,
- modifiers: Optional[Mapping[str, Any]] = None,
- ):
- # allow compatibility with libraries that
- # refer to BinaryExpression directly and pass strings
- if isinstance(operator, str):
- operator = operators.custom_op(operator)
- self._orig = (left.__hash__(), right.__hash__())
- self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
- self.left = left.self_group(against=operator)
- self.right = right.self_group(against=operator)
- self.operator = operator
- # if type is None, we get NULLTYPE, which is our _T. But I don't
- # know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
- self.negate = negate
- self._is_implicitly_boolean = operators.is_boolean(operator)
- if modifiers is None:
- self.modifiers = {}
- else:
- self.modifiers = modifiers
- @property
- def _flattened_operator_clauses(
- self,
- ) -> typing_Tuple[ColumnElement[Any], ...]:
- return (self.left, self.right)
- def __bool__(self):
- """Implement Python-side "bool" for BinaryExpression as a
- simple "identity" check for the left and right attributes,
- if the operator is "eq" or "ne". Otherwise the expression
- continues to not support "bool" like all other column expressions.
- The rationale here is so that ColumnElement objects can be hashable.
- What? Well, suppose you do this::
- c1, c2 = column("x"), column("y")
- s1 = set([c1, c2])
- We do that **a lot**, columns inside of sets is an extremely basic
- thing all over the ORM for example.
- So what happens if we do this? ::
- c1 in s1
- Hashing means it will normally use ``__hash__()`` of the object,
- but in case of hash collision, it's going to also do ``c1 == c1``
- and/or ``c1 == c2`` inside. Those operations need to return a
- True/False value. But because we override ``==`` and ``!=``, they're
- going to get a BinaryExpression. Hence we implement ``__bool__`` here
- so that these comparisons behave in this particular context mostly
- like regular object comparisons. Thankfully Python is OK with
- that! Otherwise we'd have to use special set classes for columns
- (which we used to do, decades ago).
- """
- if self.operator in (operators.eq, operators.ne):
- # this is using the eq/ne operator given int hash values,
- # rather than Operator, so that "bool" can be based on
- # identity
- return self.operator(*self._orig) # type: ignore
- else:
- raise TypeError("Boolean value of this clause is not defined")
- if typing.TYPE_CHECKING:
- def __invert__(
- self: BinaryExpression[_T],
- ) -> BinaryExpression[_T]: ...
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return self.left._from_objects + self.right._from_objects
- def _negate(self):
- if self.negate is not None:
- return BinaryExpression(
- self.left,
- self.right._negate_in_binary(self.negate, self.operator),
- self.negate,
- negate=self.operator,
- type_=self.type,
- modifiers=self.modifiers,
- )
- else:
- return self.self_group()._negate()
- class Slice(ColumnElement[Any]):
- """Represent SQL for a Python array-slice object.
- This is not a specific SQL construct at this level, but
- may be interpreted by specific dialects, e.g. PostgreSQL.
- """
- __visit_name__ = "slice"
- _traverse_internals: _TraverseInternalsType = [
- ("start", InternalTraversal.dp_clauseelement),
- ("stop", InternalTraversal.dp_clauseelement),
- ("step", InternalTraversal.dp_clauseelement),
- ]
- def __init__(self, start, stop, step, _name=None):
- self.start = coercions.expect(
- roles.ExpressionElementRole,
- start,
- name=_name,
- type_=type_api.INTEGERTYPE,
- )
- self.stop = coercions.expect(
- roles.ExpressionElementRole,
- stop,
- name=_name,
- type_=type_api.INTEGERTYPE,
- )
- self.step = coercions.expect(
- roles.ExpressionElementRole,
- step,
- name=_name,
- type_=type_api.INTEGERTYPE,
- )
- self.type = type_api.NULLTYPE
- def self_group(self, against: Optional[OperatorType] = None) -> Self:
- assert against is operator.getitem
- return self
- class IndexExpression(BinaryExpression[Any]):
- """Represent the class of expressions that are like an "index"
- operation."""
- inherit_cache = True
- class GroupedElement(DQLDMLClauseElement):
- """Represent any parenthesized expression"""
- __visit_name__ = "grouping"
- def self_group(self, against: Optional[OperatorType] = None) -> Self:
- return self
- def _ungroup(self) -> ClauseElement:
- raise NotImplementedError()
- class Grouping(GroupedElement, ColumnElement[_T]):
- """Represent a grouping within a column expression"""
- _traverse_internals: _TraverseInternalsType = [
- ("element", InternalTraversal.dp_clauseelement),
- ("type", InternalTraversal.dp_type),
- ]
- _cache_key_traversal = [
- ("element", InternalTraversal.dp_clauseelement),
- ]
- element: Union[TextClause, ClauseList, ColumnElement[_T]]
- def __init__(
- self, element: Union[TextClause, ClauseList, ColumnElement[_T]]
- ):
- self.element = element
- # nulltype assignment issue
- self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
- self._propagate_attrs = element._propagate_attrs
- def _with_binary_element_type(self, type_):
- return self.__class__(self.element._with_binary_element_type(type_))
- def _ungroup(self) -> ColumnElement[_T]:
- assert isinstance(self.element, ColumnElement)
- return self.element._ungroup()
- @util.memoized_property
- def _is_implicitly_boolean(self):
- return self.element._is_implicitly_boolean
- @util.non_memoized_property
- def _tq_label(self) -> Optional[str]:
- return (
- getattr(self.element, "_tq_label", None) or self._anon_name_label
- )
- @util.non_memoized_property
- def _proxies(self) -> List[ColumnElement[Any]]:
- if isinstance(self.element, ColumnElement):
- return [self.element]
- else:
- return []
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return self.element._from_objects
- def __getattr__(self, attr):
- return getattr(self.element, attr)
- def __getstate__(self):
- return {"element": self.element, "type": self.type}
- def __setstate__(self, state):
- self.element = state["element"]
- self.type = state["type"]
- if TYPE_CHECKING:
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> Self: ...
- class _OverrideBinds(Grouping[_T]):
- """used by cache_key->_apply_params_to_element to allow compilation /
- execution of a SQL element that's been cached, using an alternate set of
- bound parameter values.
- This is used by the ORM to swap new parameter values into expressions
- that are embedded into loader options like with_expression(),
- selectinload(). Previously, this task was accomplished using the
- .params() method which would perform a deep-copy instead. This deep
- copy proved to be too expensive for more complex expressions.
- See #11085
- """
- __visit_name__ = "override_binds"
- def __init__(
- self,
- element: ColumnElement[_T],
- bindparams: Sequence[BindParameter[Any]],
- replaces_params: Sequence[BindParameter[Any]],
- ):
- self.element = element
- self.translate = {
- k.key: v.value for k, v in zip(replaces_params, bindparams)
- }
- def _gen_cache_key(
- self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
- ) -> Optional[typing_Tuple[Any, ...]]:
- """generate a cache key for the given element, substituting its bind
- values for the translation values present."""
- existing_bps: List[BindParameter[Any]] = []
- ck = self.element._gen_cache_key(anon_map, existing_bps)
- bindparams.extend(
- (
- bp._with_value(
- self.translate[bp.key], maintain_key=True, required=False
- )
- if bp.key in self.translate
- else bp
- )
- for bp in existing_bps
- )
- return ck
- class _OverRange(Enum):
- RANGE_UNBOUNDED = 0
- RANGE_CURRENT = 1
- RANGE_UNBOUNDED = _OverRange.RANGE_UNBOUNDED
- RANGE_CURRENT = _OverRange.RANGE_CURRENT
- _IntOrRange = Union[int, _OverRange]
- class Over(ColumnElement[_T]):
- """Represent an OVER clause.
- This is a special operator against a so-called
- "window" function, as well as any aggregate function,
- which produces results relative to the result set
- itself. Most modern SQL backends now support window functions.
- """
- __visit_name__ = "over"
- _traverse_internals: _TraverseInternalsType = [
- ("element", InternalTraversal.dp_clauseelement),
- ("order_by", InternalTraversal.dp_clauseelement),
- ("partition_by", InternalTraversal.dp_clauseelement),
- ("range_", InternalTraversal.dp_plain_obj),
- ("rows", InternalTraversal.dp_plain_obj),
- ("groups", InternalTraversal.dp_plain_obj),
- ]
- order_by: Optional[ClauseList] = None
- partition_by: Optional[ClauseList] = None
- element: ColumnElement[_T]
- """The underlying expression object to which this :class:`.Over`
- object refers."""
- range_: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
- rows: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
- groups: Optional[typing_Tuple[_IntOrRange, _IntOrRange]]
- def __init__(
- self,
- element: ColumnElement[_T],
- partition_by: Optional[_ByArgument] = None,
- order_by: Optional[_ByArgument] = None,
- range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- ):
- self.element = element
- if order_by is not None:
- self.order_by = ClauseList(
- *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
- )
- if partition_by is not None:
- self.partition_by = ClauseList(
- *util.to_list(partition_by),
- _literal_as_text_role=roles.ByOfRole,
- )
- if sum(bool(item) for item in (range_, rows, groups)) > 1:
- raise exc.ArgumentError(
- "only one of 'rows', 'range_', or 'groups' may be provided"
- )
- else:
- self.range_ = self._interpret_range(range_) if range_ else None
- self.rows = self._interpret_range(rows) if rows else None
- self.groups = self._interpret_range(groups) if groups else None
- def __reduce__(self):
- return self.__class__, (
- self.element,
- self.partition_by,
- self.order_by,
- self.range_,
- self.rows,
- self.groups,
- )
- def _interpret_range(
- self,
- range_: typing_Tuple[Optional[_IntOrRange], Optional[_IntOrRange]],
- ) -> typing_Tuple[_IntOrRange, _IntOrRange]:
- if not isinstance(range_, tuple) or len(range_) != 2:
- raise exc.ArgumentError("2-tuple expected for range/rows")
- r0, r1 = range_
- lower: _IntOrRange
- upper: _IntOrRange
- if r0 is None:
- lower = RANGE_UNBOUNDED
- elif isinstance(r0, _OverRange):
- lower = r0
- else:
- try:
- lower = int(r0)
- except ValueError as err:
- raise exc.ArgumentError(
- "Integer or None expected for range value"
- ) from err
- else:
- if lower == 0:
- lower = RANGE_CURRENT
- if r1 is None:
- upper = RANGE_UNBOUNDED
- elif isinstance(r1, _OverRange):
- upper = r1
- else:
- try:
- upper = int(r1)
- except ValueError as err:
- raise exc.ArgumentError(
- "Integer or None expected for range value"
- ) from err
- else:
- if upper == 0:
- upper = RANGE_CURRENT
- return lower, upper
- if not TYPE_CHECKING:
- @util.memoized_property
- def type(self) -> TypeEngine[_T]: # noqa: A001
- return self.element.type
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return list(
- itertools.chain(
- *[
- c._from_objects
- for c in (self.element, self.partition_by, self.order_by)
- if c is not None
- ]
- )
- )
- class WithinGroup(ColumnElement[_T]):
- """Represent a WITHIN GROUP (ORDER BY) clause.
- This is a special operator against so-called
- "ordered set aggregate" and "hypothetical
- set aggregate" functions, including ``percentile_cont()``,
- ``rank()``, ``dense_rank()``, etc.
- It's supported only by certain database backends, such as PostgreSQL,
- Oracle Database and MS SQL Server.
- The :class:`.WithinGroup` construct extracts its type from the
- method :meth:`.FunctionElement.within_group_type`. If this returns
- ``None``, the function's ``.type`` is used.
- """
- __visit_name__ = "withingroup"
- _traverse_internals: _TraverseInternalsType = [
- ("element", InternalTraversal.dp_clauseelement),
- ("order_by", InternalTraversal.dp_clauseelement),
- ]
- order_by: Optional[ClauseList] = None
- def __init__(
- self,
- element: Union[FunctionElement[_T], FunctionFilter[_T]],
- *order_by: _ColumnExpressionArgument[Any],
- ):
- self.element = element
- if order_by is not None:
- self.order_by = ClauseList(
- *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
- )
- def __reduce__(self):
- return self.__class__, (self.element,) + (
- tuple(self.order_by) if self.order_by is not None else ()
- )
- def over(
- self,
- *,
- partition_by: Optional[_ByArgument] = None,
- order_by: Optional[_ByArgument] = None,
- rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- ) -> Over[_T]:
- """Produce an OVER clause against this :class:`.WithinGroup`
- construct.
- This function has the same signature as that of
- :meth:`.FunctionElement.over`.
- """
- return Over(
- self,
- partition_by=partition_by,
- order_by=order_by,
- range_=range_,
- rows=rows,
- groups=groups,
- )
- @overload
- def filter(self) -> Self: ...
- @overload
- def filter(
- self,
- __criterion0: _ColumnExpressionArgument[bool],
- *criterion: _ColumnExpressionArgument[bool],
- ) -> FunctionFilter[_T]: ...
- def filter(
- self, *criterion: _ColumnExpressionArgument[bool]
- ) -> Union[Self, FunctionFilter[_T]]:
- """Produce a FILTER clause against this function."""
- if not criterion:
- return self
- return FunctionFilter(self, *criterion)
- if not TYPE_CHECKING:
- @util.memoized_property
- def type(self) -> TypeEngine[_T]: # noqa: A001
- wgt = self.element.within_group_type(self)
- if wgt is not None:
- return wgt
- else:
- return self.element.type
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return list(
- itertools.chain(
- *[
- c._from_objects
- for c in (self.element, self.order_by)
- if c is not None
- ]
- )
- )
- class FunctionFilter(Generative, ColumnElement[_T]):
- """Represent a function FILTER clause.
- This is a special operator against aggregate and window functions,
- which controls which rows are passed to it.
- It's supported only by certain database backends.
- Invocation of :class:`.FunctionFilter` is via
- :meth:`.FunctionElement.filter`::
- func.count(1).filter(True)
- .. seealso::
- :meth:`.FunctionElement.filter`
- """
- __visit_name__ = "funcfilter"
- _traverse_internals: _TraverseInternalsType = [
- ("func", InternalTraversal.dp_clauseelement),
- ("criterion", InternalTraversal.dp_clauseelement),
- ]
- criterion: Optional[ColumnElement[bool]] = None
- def __init__(
- self,
- func: Union[FunctionElement[_T], WithinGroup[_T]],
- *criterion: _ColumnExpressionArgument[bool],
- ):
- self.func = func
- self.filter.non_generative(self, *criterion) # type: ignore
- @_generative
- def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
- """Produce an additional FILTER against the function.
- This method adds additional criteria to the initial criteria
- set up by :meth:`.FunctionElement.filter`.
- Multiple criteria are joined together at SQL render time
- via ``AND``.
- """
- for crit in list(criterion):
- crit = coercions.expect(roles.WhereHavingRole, crit)
- if self.criterion is not None:
- self.criterion = self.criterion & crit
- else:
- self.criterion = crit
- return self
- def over(
- self,
- partition_by: Optional[
- Union[
- Iterable[_ColumnExpressionArgument[Any]],
- _ColumnExpressionArgument[Any],
- ]
- ] = None,
- order_by: Optional[
- Union[
- Iterable[_ColumnExpressionArgument[Any]],
- _ColumnExpressionArgument[Any],
- ]
- ] = None,
- range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
- ) -> Over[_T]:
- """Produce an OVER clause against this filtered function.
- Used against aggregate or so-called "window" functions,
- for database backends that support window functions.
- The expression::
- func.rank().filter(MyClass.y > 5).over(order_by="x")
- is shorthand for::
- from sqlalchemy import over, funcfilter
- over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
- See :func:`_expression.over` for a full description.
- """
- return Over(
- self,
- partition_by=partition_by,
- order_by=order_by,
- range_=range_,
- rows=rows,
- groups=groups,
- )
- def within_group(
- self, *order_by: _ColumnExpressionArgument[Any]
- ) -> WithinGroup[_T]:
- """Produce a WITHIN GROUP (ORDER BY expr) clause against
- this function.
- """
- return WithinGroup(self, *order_by)
- def within_group_type(
- self, within_group: WithinGroup[_T]
- ) -> Optional[TypeEngine[_T]]:
- return None
- def self_group(
- self, against: Optional[OperatorType] = None
- ) -> Union[Self, Grouping[_T]]:
- if operators.is_precedent(operators.filter_op, against):
- return Grouping(self)
- else:
- return self
- if not TYPE_CHECKING:
- @util.memoized_property
- def type(self) -> TypeEngine[_T]: # noqa: A001
- return self.func.type
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return list(
- itertools.chain(
- *[
- c._from_objects
- for c in (self.func, self.criterion)
- if c is not None
- ]
- )
- )
- class NamedColumn(KeyedColumnElement[_T]):
- is_literal = False
- table: Optional[FromClause] = None
- name: str
- key: str
- def _compare_name_for_result(self, other):
- return (hasattr(other, "name") and self.name == other.name) or (
- hasattr(other, "_label") and self._label == other._label
- )
- @util.ro_memoized_property
- def description(self) -> str:
- return self.name
- @HasMemoized.memoized_attribute
- def _tq_key_label(self) -> Optional[str]:
- """table qualified label based on column key.
- for table-bound columns this is <tablename>_<column key/proxy key>;
- all other expressions it resolves to key/proxy key.
- """
- proxy_key = self._proxy_key
- if proxy_key and proxy_key != self.name:
- return self._gen_tq_label(proxy_key)
- else:
- return self._tq_label
- @HasMemoized.memoized_attribute
- def _tq_label(self) -> Optional[str]:
- """table qualified label based on column name.
- for table-bound columns this is <tablename>_<columnname>; all other
- expressions it resolves to .name.
- """
- return self._gen_tq_label(self.name)
- @HasMemoized.memoized_attribute
- def _render_label_in_columns_clause(self):
- return True
- @HasMemoized.memoized_attribute
- def _non_anon_label(self):
- return self.name
- def _gen_tq_label(
- self, name: str, dedupe_on_key: bool = True
- ) -> Optional[str]:
- return name
- def _bind_param(
- self,
- operator: OperatorType,
- obj: Any,
- type_: Optional[TypeEngine[_T]] = None,
- expanding: bool = False,
- ) -> BindParameter[_T]:
- return BindParameter(
- self.key,
- obj,
- _compared_to_operator=operator,
- _compared_to_type=self.type,
- type_=type_,
- unique=True,
- expanding=expanding,
- )
- def _make_proxy(
- self,
- selectable: FromClause,
- *,
- primary_key: ColumnSet,
- foreign_keys: Set[KeyedColumnElement[Any]],
- name: Optional[str] = None,
- key: Optional[str] = None,
- name_is_truncatable: bool = False,
- compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
- disallow_is_literal: bool = False,
- **kw: Any,
- ) -> typing_Tuple[str, ColumnClause[_T]]:
- c = ColumnClause(
- (
- coercions.expect(roles.TruncatedLabelRole, name or self.name)
- if name_is_truncatable
- else (name or self.name)
- ),
- type_=self.type,
- _selectable=selectable,
- is_literal=False,
- )
- c._propagate_attrs = selectable._propagate_attrs
- if name is None:
- c.key = self.key
- if compound_select_cols:
- c._proxies = list(compound_select_cols)
- else:
- c._proxies = [self]
- if selectable._is_clone_of is not None:
- c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
- return c.key, c
- _PS = ParamSpec("_PS")
- class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
- """Represents a column label (AS).
- Represent a label, as typically applied to any column-level
- element using the ``AS`` sql keyword.
- """
- __visit_name__ = "label"
- _traverse_internals: _TraverseInternalsType = [
- ("name", InternalTraversal.dp_anon_name),
- ("type", InternalTraversal.dp_type),
- ("_element", InternalTraversal.dp_clauseelement),
- ]
- _cache_key_traversal = [
- ("name", InternalTraversal.dp_anon_name),
- ("_element", InternalTraversal.dp_clauseelement),
- ]
- _element: ColumnElement[_T]
- name: str
- def __init__(
- self,
- name: Optional[str],
- element: _ColumnExpressionArgument[_T],
- type_: Optional[_TypeEngineArgument[_T]] = None,
- ):
- orig_element = element
- element = coercions.expect(
- roles.ExpressionElementRole,
- element,
- apply_propagate_attrs=self,
- )
- while isinstance(element, Label):
- # TODO: this is only covered in test_text.py, but nothing
- # fails if it's removed. determine rationale
- element = element.element
- if name:
- self.name = name
- else:
- self.name = _anonymous_label.safe_construct(
- id(self), getattr(element, "name", "anon")
- )
- if isinstance(orig_element, Label):
- # TODO: no coverage for this block, again would be in
- # test_text.py where the resolve_label concept is important
- self._resolve_label = orig_element._label
- self.key = self._tq_label = self._tq_key_label = self.name
- self._element = element
- self.type = (
- type_api.to_instance(type_)
- if type_ is not None
- else self._element.type
- )
- self._proxies = [element]
- def __reduce__(self):
- return self.__class__, (self.name, self._element, self.type)
- @HasMemoized.memoized_attribute
- def _render_label_in_columns_clause(self):
- return True
- def _bind_param(self, operator, obj, type_=None, expanding=False):
- return BindParameter(
- None,
- obj,
- _compared_to_operator=operator,
- type_=type_,
- _compared_to_type=self.type,
- unique=True,
- expanding=expanding,
- )
- @util.memoized_property
- def _is_implicitly_boolean(self):
- return self.element._is_implicitly_boolean
- @HasMemoized.memoized_attribute
- def _allow_label_resolve(self):
- return self.element._allow_label_resolve
- @property
- def _order_by_label_element(self):
- return self
- @HasMemoized.memoized_attribute
- def element(self) -> ColumnElement[_T]:
- return self._element.self_group(against=operators.as_)
- def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
- return self._apply_to_inner(self._element.self_group, against=against)
- def _negate(self):
- return self._apply_to_inner(self._element._negate)
- def _apply_to_inner(
- self,
- fn: Callable[_PS, ColumnElement[_T]],
- *arg: _PS.args,
- **kw: _PS.kwargs,
- ) -> Label[_T]:
- sub_element = fn(*arg, **kw)
- if sub_element is not self._element:
- return Label(self.name, sub_element, type_=self.type)
- else:
- return self
- @property
- def primary_key(self): # type: ignore[override]
- return self.element.primary_key
- @property
- def foreign_keys(self): # type: ignore[override]
- return self.element.foreign_keys
- def _copy_internals(
- self,
- *,
- clone: _CloneCallableType = _clone,
- anonymize_labels: bool = False,
- **kw: Any,
- ) -> None:
- self._reset_memoizations()
- self._element = clone(self._element, **kw)
- if anonymize_labels:
- self.name = _anonymous_label.safe_construct(
- id(self), getattr(self.element, "name", "anon")
- )
- self.key = self._tq_label = self._tq_key_label = self.name
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return self.element._from_objects
- def _make_proxy(
- self,
- selectable: FromClause,
- *,
- primary_key: ColumnSet,
- foreign_keys: Set[KeyedColumnElement[Any]],
- name: Optional[str] = None,
- compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
- **kw: Any,
- ) -> typing_Tuple[str, ColumnClause[_T]]:
- name = self.name if not name else name
- key, e = self.element._make_proxy(
- selectable,
- name=name,
- disallow_is_literal=True,
- name_is_truncatable=isinstance(name, _truncated_label),
- compound_select_cols=compound_select_cols,
- primary_key=primary_key,
- foreign_keys=foreign_keys,
- )
- # there was a note here to remove this assertion, which was here
- # to determine if we later could support a use case where
- # the key and name of a label are separate. But I don't know what
- # that case was. For now, this is an unexpected case that occurs
- # when a label name conflicts with other columns and select()
- # is attempting to disambiguate an explicit label, which is not what
- # the user would want. See issue #6090.
- if key != self.name and not isinstance(self.name, _anonymous_label):
- raise exc.InvalidRequestError(
- "Label name %s is being renamed to an anonymous label due "
- "to disambiguation "
- "which is not supported right now. Please use unique names "
- "for explicit labels." % (self.name)
- )
- e._propagate_attrs = selectable._propagate_attrs
- e._proxies.append(self)
- if self.type is not None:
- e.type = self.type
- return self.key, e
- class ColumnClause(
- roles.DDLReferredColumnRole,
- roles.LabeledColumnExprRole[_T],
- roles.StrAsPlainColumnRole,
- Immutable,
- NamedColumn[_T],
- ):
- """Represents a column expression from any textual string.
- The :class:`.ColumnClause`, a lightweight analogue to the
- :class:`_schema.Column` class, is typically invoked using the
- :func:`_expression.column` function, as in::
- from sqlalchemy import column
- id, name = column("id"), column("name")
- stmt = select(id, name).select_from("user")
- The above statement would produce SQL like:
- .. sourcecode:: sql
- SELECT id, name FROM user
- :class:`.ColumnClause` is the immediate superclass of the schema-specific
- :class:`_schema.Column` object. While the :class:`_schema.Column`
- class has all the
- same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
- class is usable by itself in those cases where behavioral requirements
- are limited to simple SQL expression generation. The object has none of
- the associations with schema-level metadata or with execution-time
- behavior that :class:`_schema.Column` does,
- so in that sense is a "lightweight"
- version of :class:`_schema.Column`.
- Full details on :class:`.ColumnClause` usage is at
- :func:`_expression.column`.
- .. seealso::
- :func:`_expression.column`
- :class:`_schema.Column`
- """
- table: Optional[FromClause]
- is_literal: bool
- __visit_name__ = "column"
- _traverse_internals: _TraverseInternalsType = [
- ("name", InternalTraversal.dp_anon_name),
- ("type", InternalTraversal.dp_type),
- ("table", InternalTraversal.dp_clauseelement),
- ("is_literal", InternalTraversal.dp_boolean),
- ]
- onupdate: Optional[DefaultGenerator] = None
- default: Optional[DefaultGenerator] = None
- server_default: Optional[FetchedValue] = None
- server_onupdate: Optional[FetchedValue] = None
- _is_multiparam_column = False
- @property
- def _is_star(self): # type: ignore[override]
- return self.is_literal and self.name == "*"
- def __init__(
- self,
- text: str,
- type_: Optional[_TypeEngineArgument[_T]] = None,
- is_literal: bool = False,
- _selectable: Optional[FromClause] = None,
- ):
- self.key = self.name = text
- self.table = _selectable
- # if type is None, we get NULLTYPE, which is our _T. But I don't
- # know how to get the overloads to express that correctly
- self.type = type_api.to_instance(type_) # type: ignore
- self.is_literal = is_literal
- def get_children(self, *, column_tables=False, **kw):
- # override base get_children() to not return the Table
- # or selectable that is parent to this column. Traversals
- # expect the columns of tables and subqueries to be leaf nodes.
- return []
- @property
- def entity_namespace(self):
- if self.table is not None:
- return self.table.entity_namespace
- else:
- return super().entity_namespace
- def _clone(self, detect_subquery_cols=False, **kw):
- if (
- detect_subquery_cols
- and self.table is not None
- and self.table._is_subquery
- ):
- clone = kw.pop("clone")
- table = clone(self.table, **kw)
- new = table.c.corresponding_column(self)
- return new
- return super()._clone(**kw)
- @HasMemoized_ro_memoized_attribute
- def _from_objects(self) -> List[FromClause]:
- t = self.table
- if t is not None:
- return [t]
- else:
- return []
- @HasMemoized.memoized_attribute
- def _render_label_in_columns_clause(self):
- return self.table is not None
- @property
- def _ddl_label(self):
- return self._gen_tq_label(self.name, dedupe_on_key=False)
- def _compare_name_for_result(self, other):
- if (
- self.is_literal
- or self.table is None
- or self.table._is_textual
- or not hasattr(other, "proxy_set")
- or (
- isinstance(other, ColumnClause)
- and (
- other.is_literal
- or other.table is None
- or other.table._is_textual
- )
- )
- ):
- return (hasattr(other, "name") and self.name == other.name) or (
- hasattr(other, "_tq_label")
- and self._tq_label == other._tq_label
- )
- else:
- return other.proxy_set.intersection(self.proxy_set)
- def _gen_tq_label(
- self, name: str, dedupe_on_key: bool = True
- ) -> Optional[str]:
- """generate table-qualified label
- for a table-bound column this is <tablename>_<columnname>.
- used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
- as well as the .columns collection on a Join object.
- """
- label: str
- t = self.table
- if self.is_literal:
- return None
- elif t is not None and is_named_from_clause(t):
- if has_schema_attr(t) and t.schema:
- label = t.schema.replace(".", "_") + "_" + t.name + "_" + name
- else:
- assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
- label = t.name + "_" + name
- # propagate name quoting rules for labels.
- if is_quoted_name(name) and name.quote is not None:
- if is_quoted_name(label):
- label.quote = name.quote
- else:
- label = quoted_name(label, name.quote)
- elif is_quoted_name(t.name) and t.name.quote is not None:
- # can't get this situation to occur, so let's
- # assert false on it for now
- assert not isinstance(label, quoted_name)
- label = quoted_name(label, t.name.quote)
- if dedupe_on_key:
- # ensure the label name doesn't conflict with that of an
- # existing column. note that this implies that any Column
- # must **not** set up its _label before its parent table has
- # all of its other Column objects set up. There are several
- # tables in the test suite which will fail otherwise; example:
- # table "owner" has columns "name" and "owner_name". Therefore
- # column owner.name cannot use the label "owner_name", it has
- # to be "owner_name_1".
- if label in t.c:
- _label = label
- counter = 1
- while _label in t.c:
- _label = label + "_" + str(counter)
- counter += 1
- label = _label
- return coercions.expect(roles.TruncatedLabelRole, label)
- else:
- return name
- def _make_proxy(
- self,
- selectable: FromClause,
- *,
- primary_key: ColumnSet,
- foreign_keys: Set[KeyedColumnElement[Any]],
- name: Optional[str] = None,
- key: Optional[str] = None,
- name_is_truncatable: bool = False,
- compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
- disallow_is_literal: bool = False,
- **kw: Any,
- ) -> typing_Tuple[str, ColumnClause[_T]]:
- # the "is_literal" flag normally should never be propagated; a proxied
- # column is always a SQL identifier and never the actual expression
- # being evaluated. however, there is a case where the "is_literal" flag
- # might be used to allow the given identifier to have a fixed quoting
- # pattern already, so maintain the flag for the proxy unless a
- # :class:`.Label` object is creating the proxy. See [ticket:4730].
- is_literal = (
- not disallow_is_literal
- and self.is_literal
- and (
- # note this does not accommodate for quoted_name differences
- # right now
- name is None
- or name == self.name
- )
- )
- c = self._constructor(
- (
- coercions.expect(roles.TruncatedLabelRole, name or self.name)
- if name_is_truncatable
- else (name or self.name)
- ),
- type_=self.type,
- _selectable=selectable,
- is_literal=is_literal,
- )
- c._propagate_attrs = selectable._propagate_attrs
- if name is None:
- c.key = self.key
- if compound_select_cols:
- c._proxies = list(compound_select_cols)
- else:
- c._proxies = [self]
- if selectable._is_clone_of is not None:
- c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
- return c.key, c
- class TableValuedColumn(NamedColumn[_T]):
- __visit_name__ = "table_valued_column"
- _traverse_internals: _TraverseInternalsType = [
- ("name", InternalTraversal.dp_anon_name),
- ("type", InternalTraversal.dp_type),
- ("scalar_alias", InternalTraversal.dp_clauseelement),
- ]
- def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
- self.scalar_alias = scalar_alias
- self.key = self.name = scalar_alias.name
- self.type = type_
- def _copy_internals(
- self, clone: _CloneCallableType = _clone, **kw: Any
- ) -> None:
- self.scalar_alias = clone(self.scalar_alias, **kw)
- self.key = self.name = self.scalar_alias.name
- @util.ro_non_memoized_property
- def _from_objects(self) -> List[FromClause]:
- return [self.scalar_alias]
- class CollationClause(ColumnElement[str]):
- __visit_name__ = "collation"
- _traverse_internals: _TraverseInternalsType = [
- ("collation", InternalTraversal.dp_string)
- ]
- @classmethod
- @util.preload_module("sqlalchemy.sql.sqltypes")
- def _create_collation_expression(
- cls, expression: _ColumnExpressionArgument[str], collation: str
- ) -> BinaryExpression[str]:
- sqltypes = util.preloaded.sql_sqltypes
- expr = coercions.expect(roles.ExpressionElementRole[str], expression)
- if expr.type._type_affinity is sqltypes.String:
- collate_type = expr.type._with_collation(collation)
- else:
- collate_type = expr.type
- return BinaryExpression(
- expr,
- CollationClause(collation),
- operators.collate,
- type_=collate_type,
- )
- def __init__(self, collation):
- self.collation = collation
- class _IdentifiedClause(Executable, ClauseElement):
- __visit_name__ = "identified"
- def __init__(self, ident):
- self.ident = ident
- class SavepointClause(_IdentifiedClause):
- __visit_name__ = "savepoint"
- inherit_cache = False
- class RollbackToSavepointClause(_IdentifiedClause):
- __visit_name__ = "rollback_to_savepoint"
- inherit_cache = False
- class ReleaseSavepointClause(_IdentifiedClause):
- __visit_name__ = "release_savepoint"
- inherit_cache = False
- class quoted_name(util.MemoizedSlots, str):
- """Represent a SQL identifier combined with quoting preferences.
- :class:`.quoted_name` is a Python unicode/str subclass which
- represents a particular identifier name along with a
- ``quote`` flag. This ``quote`` flag, when set to
- ``True`` or ``False``, overrides automatic quoting behavior
- for this identifier in order to either unconditionally quote
- or to not quote the name. If left at its default of ``None``,
- quoting behavior is applied to the identifier on a per-backend basis
- based on an examination of the token itself.
- A :class:`.quoted_name` object with ``quote=True`` is also
- prevented from being modified in the case of a so-called
- "name normalize" option. Certain database backends, such as
- Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
- as uppercase. The SQLAlchemy dialects for these backends
- convert from SQLAlchemy's lower-case-means-insensitive convention
- to the upper-case-means-insensitive conventions of those backends.
- The ``quote=True`` flag here will prevent this conversion from occurring
- to support an identifier that's quoted as all lower case against
- such a backend.
- The :class:`.quoted_name` object is normally created automatically
- when specifying the name for key schema constructs such as
- :class:`_schema.Table`, :class:`_schema.Column`, and others.
- The class can also be
- passed explicitly as the name to any function that receives a name which
- can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
- method with
- an unconditionally quoted name::
- from sqlalchemy import create_engine
- from sqlalchemy import inspect
- from sqlalchemy.sql import quoted_name
- engine = create_engine("oracle+oracledb://some_dsn")
- print(inspect(engine).has_table(quoted_name("some_table", True)))
- The above logic will run the "has table" logic against the Oracle Database
- backend, passing the name exactly as ``"some_table"`` without converting to
- upper case.
- .. versionchanged:: 1.2 The :class:`.quoted_name` construct is now
- importable from ``sqlalchemy.sql``, in addition to the previous
- location of ``sqlalchemy.sql.elements``.
- """
- __slots__ = "quote", "lower", "upper"
- quote: Optional[bool]
- @overload
- @classmethod
- def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
- @overload
- @classmethod
- def construct(cls, value: None, quote: Optional[bool]) -> None: ...
- @classmethod
- def construct(
- cls, value: Optional[str], quote: Optional[bool]
- ) -> Optional[quoted_name]:
- if value is None:
- return None
- else:
- return quoted_name(value, quote)
- def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
- assert (
- value is not None
- ), "use quoted_name.construct() for None passthrough"
- if isinstance(value, cls) and (quote is None or value.quote == quote):
- return value
- self = super().__new__(cls, value)
- self.quote = quote
- return self
- def __reduce__(self):
- return quoted_name, (str(self), self.quote)
- def _memoized_method_lower(self):
- if self.quote:
- return self
- else:
- return str(self).lower()
- def _memoized_method_upper(self):
- if self.quote:
- return self
- else:
- return str(self).upper()
- def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
- """locate Column objects within the given expression."""
- cols: Set[ColumnClause[Any]] = set()
- traverse(clause, {}, {"column": cols.add})
- return cols
- def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
- for a in args:
- if not a.type._isnull:
- return a.type
- else:
- return type_api.NULLTYPE # type: ignore
- def _corresponding_column_or_error(fromclause, column, require_embedded=False):
- c = fromclause.corresponding_column(
- column, require_embedded=require_embedded
- )
- if c is None:
- raise exc.InvalidRequestError(
- "Given column '%s', attached to table '%s', "
- "failed to locate a corresponding column from table '%s'"
- % (column, getattr(column, "table", None), fromclause.description)
- )
- return c
- class _memoized_property_but_not_nulltype(
- util.memoized_property["TypeEngine[_T]"]
- ):
- """memoized property, but dont memoize NullType"""
- def __get__(self, obj, cls):
- if obj is None:
- return self
- result = self.fget(obj)
- if not result._isnull:
- obj.__dict__[self.__name__] = result
- return result
- class AnnotatedColumnElement(Annotated):
- _Annotated__element: ColumnElement[Any]
- def __init__(self, element, values):
- Annotated.__init__(self, element, values)
- for attr in (
- "comparator",
- "_proxy_key",
- "_tq_key_label",
- "_tq_label",
- "_non_anon_label",
- "type",
- ):
- self.__dict__.pop(attr, None)
- for attr in ("name", "key", "table"):
- if self.__dict__.get(attr, False) is None:
- self.__dict__.pop(attr)
- def _with_annotations(self, values):
- clone = super()._with_annotations(values)
- for attr in (
- "comparator",
- "_proxy_key",
- "_tq_key_label",
- "_tq_label",
- "_non_anon_label",
- ):
- clone.__dict__.pop(attr, None)
- return clone
- @util.memoized_property
- def name(self):
- """pull 'name' from parent, if not present"""
- return self._Annotated__element.name
- @_memoized_property_but_not_nulltype
- def type(self):
- """pull 'type' from parent and don't cache if null.
- type is routinely changed on existing columns within the
- mapped_column() initialization process, and "type" is also consulted
- during the creation of SQL expressions. Therefore it can change after
- it was already retrieved. At the same time we don't want annotated
- objects having overhead when expressions are produced, so continue
- to memoize, but only when we have a non-null type.
- """
- return self._Annotated__element.type
- @util.memoized_property
- def table(self):
- """pull 'table' from parent, if not present"""
- return self._Annotated__element.table
- @util.memoized_property
- def key(self):
- """pull 'key' from parent, if not present"""
- return self._Annotated__element.key
- @util.memoized_property
- def info(self) -> _InfoType:
- if TYPE_CHECKING:
- assert isinstance(self._Annotated__element, Column)
- return self._Annotated__element.info
- @util.memoized_property
- def _anon_name_label(self) -> str:
- return self._Annotated__element._anon_name_label
- class _truncated_label(quoted_name):
- """A unicode subclass used to identify symbolic "
- "names that may require truncation."""
- __slots__ = ()
- def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
- quote = getattr(value, "quote", quote)
- # return super(_truncated_label, cls).__new__(cls, value, quote, True)
- return super().__new__(cls, value, quote)
- def __reduce__(self) -> Any:
- return self.__class__, (str(self), self.quote)
- def apply_map(self, map_: Mapping[str, Any]) -> str:
- return self
- class conv(_truncated_label):
- """Mark a string indicating that a name has already been converted
- by a naming convention.
- This is a string subclass that indicates a name that should not be
- subject to any further naming conventions.
- E.g. when we create a :class:`.Constraint` using a naming convention
- as follows::
- m = MetaData(
- naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
- )
- t = Table(
- "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
- )
- The name of the above constraint will be rendered as ``"ck_t_x5"``.
- That is, the existing name ``x5`` is used in the naming convention as the
- ``constraint_name`` token.
- In some situations, such as in migration scripts, we may be rendering
- the above :class:`.CheckConstraint` with a name that's already been
- converted. In order to make sure the name isn't double-modified, the
- new name is applied using the :func:`_schema.conv` marker. We can
- use this explicitly as follows::
- m = MetaData(
- naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
- )
- t = Table(
- "t",
- m,
- Column("x", Integer),
- CheckConstraint("x > 5", name=conv("ck_t_x5")),
- )
- Where above, the :func:`_schema.conv` marker indicates that the constraint
- name here is final, and the name will render as ``"ck_t_x5"`` and not
- ``"ck_t_ck_t_x5"``
- .. seealso::
- :ref:`constraint_naming_conventions`
- """
- __slots__ = ()
- # for backwards compatibility in case
- # someone is re-implementing the
- # _truncated_identifier() sequence in a custom
- # compiler
- _generated_label = _truncated_label
- class _anonymous_label(_truncated_label):
- """A unicode subclass used to identify anonymously
- generated names."""
- __slots__ = ()
- @classmethod
- def safe_construct(
- cls,
- seed: int,
- body: str,
- enclosing_label: Optional[str] = None,
- sanitize_key: bool = False,
- ) -> _anonymous_label:
- # need to escape chars that interfere with format
- # strings in any case, issue #8724
- body = re.sub(r"[%\(\) \$]+", "_", body)
- if sanitize_key:
- # sanitize_key is then an extra step used by BindParameter
- body = body.strip("_")
- label = "%%(%d %s)s" % (seed, body.replace("%", "%%"))
- if enclosing_label:
- label = "%s%s" % (enclosing_label, label)
- return _anonymous_label(label)
- def __add__(self, other):
- if "%" in other and not isinstance(other, _anonymous_label):
- other = str(other).replace("%", "%%")
- else:
- other = str(other)
- return _anonymous_label(
- quoted_name(
- str.__add__(self, other),
- self.quote,
- )
- )
- def __radd__(self, other):
- if "%" in other and not isinstance(other, _anonymous_label):
- other = str(other).replace("%", "%%")
- else:
- other = str(other)
- return _anonymous_label(
- quoted_name(
- str.__add__(other, self),
- self.quote,
- )
- )
- def apply_map(self, map_):
- if self.quote is not None:
- # preserve quoting only if necessary
- return quoted_name(self % map_, self.quote)
- else:
- # else skip the constructor call
- return self % map_
|