D6iKddlZddlZddlmZmZddlmZddlZddlm Z ddl m Z m Z ddl mZdd lmZdd Zdd Zdd Zd ZdZdZdej.j0j2fdZdZdZdZddZdZy)N)MappingSequence)copy)make_url)OperationalErrorProgrammingError) starts_with)quotecx|j||dzjd|dzjd|dzS)a7 Escape the string parameter used in SQL LIKE expressions. :: from sqlalchemy_utils import escape_like query = session.query(User).filter( User.name.ilike(escape_like('John')) ) :param string: a string to escape :param escape_char: escape character r %_)replace)string escape_chars f/home/azureuser/techstart-app/venv/lib/python3.12/site-packages/sqlalchemy_utils/functions/database.py escape_likers=$ {K!O4 kC' ( kC' (c|rd}ntj}t|trAtjj dt j|jDSt|tr |d|dSt|tr#tjjd|DSt|ttfr|t|S|S)a Convert python data structures to PostgreSQL specific SQLAlchemy JSON constructs. This function is extremly useful if you need to build PostgreSQL JSON on python side. .. note:: This function needs PostgreSQL >= 9.4 Scalars are converted to to_json SQLAlchemy function objects :: json_sql(1) # Equals SQL: to_json(1) json_sql('a') # to_json('a') Mappings are converted to json_build_object constructs :: json_sql({'a': 'c', '2': 5}) # json_build_object('a', 'c', '2', 5) Sequences (other than strings) are converted to json_build_array constructs :: json_sql([1, 2, 3]) # json_build_array(1, 2, 3) You can also nest these data structures :: json_sql({'a': [1, 2, 3]}) # json_build_object('a', json_build_array[1, 2, 3]) :param value: value to be converted to SQLAlchemy PostgreSQL function constructs cftjjtj|SN)safuncto_jsontextas rscalar_convertz json_sql..scalar_convertTs77??2771:. .rc36K|]}t|dywF)scalars_to_jsonNjson_sql.0vs r zjson_sql..[s"E22'c36K|]}t|dywr!r#r%s rr(zjson_sql..ds @Qhq%00 @r))rr isinstancerrjson_build_object itertoolschainitemsstrrjson_build_arrayintfloat)valuer"rs rr$r$&sX /%!ww(("%++-8  E3 %l++ E8 $ww'' @% @   EC< (c%j)) Lrc|rd}ntj}t|trAtjj dt j|jDSt|tr |d|dSt|tr#tjjd|DSt|ttfr|t|S|S)a" Convert python data structures to PostgreSQL specific SQLAlchemy JSONB constructs. This function is extremly useful if you need to build PostgreSQL JSONB on python side. .. note:: This function needs PostgreSQL >= 9.4 Scalars are converted to to_jsonb SQLAlchemy function objects :: jsonb_sql(1) # Equals SQL: to_jsonb(1) jsonb_sql('a') # to_jsonb('a') Mappings are converted to jsonb_build_object constructs :: jsonb_sql({'a': 'c', '2': 5}) # jsonb_build_object('a', 'c', '2', 5) Sequences (other than strings) converted to jsonb_build_array constructs :: jsonb_sql([1, 2, 3]) # jsonb_build_array(1, 2, 3) You can also nest these data structures :: jsonb_sql({'a': [1, 2, 3]}) # jsonb_build_object('a', jsonb_build_array[1, 2, 3]) :param value: value to be converted to SQLAlchemy PostgreSQL function constructs :boolean jsonbb: Flag to alternatively convert the return with a to_jsonb construct cftjjtj|Sr)rrto_jsonbrrs rrz!jsonb_sql..scalar_converts77##BGGAJ/ /rc36K|]}t|dywF)scalars_to_jsonbN jsonb_sqlr%s rr(zjsonb_sql..s"!e44r)r*c36K|]}t|dywr:r<r%s rr(zjsonb_sql..s BqiE22 Br))rrr,rrjsonb_build_objectr.r/r0r1rjsonb_build_arrayr3r4)r5r;rs rr=r=ks\ 0%!ww))"%++-8  E3 %l++ E8 $ww(( BE B   EC< (c%j)) Lrc|j}t|tjst d|z|j j j}t|tjr$t|j jn|g|xr t|xstfd|jDS)a Return whether or not given column or the columns of given foreign key constraint have an index. A column has an index if it has a single column index or it is the first column in compound column index. A foreign key constraint has an index if the constraint columns are the first columns in compound column index. :param column_or_constraint: SQLAlchemy Column object or SA ForeignKeyConstraint object .. versionadded: 0.26.2 .. versionchanged: 0.30.18 Added support for foreign key constaints. :: from sqlalchemy_utils import has_index class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) title = sa.Column(sa.String(100)) is_published = sa.Column(sa.Boolean, index=True) is_deleted = sa.Column(sa.Boolean) is_archived = sa.Column(sa.Boolean) __table_args__ = ( sa.Index('my_index', is_deleted, is_archived), ) table = Article.__table__ has_index(table.c.is_published) # True has_index(table.c.is_deleted) # True has_index(table.c.is_archived) # False Also supports primary key indexes :: from sqlalchemy_utils import has_index class ArticleTranslation(Base): __tablename__ = 'article_translation' id = sa.Column(sa.Integer, primary_key=True) locale = sa.Column(sa.String(10), primary_key=True) title = sa.Column(sa.String(100)) table = ArticleTranslation.__table__ has_index(table.c.locale) # False has_index(table.c.id) # True This function supports foreign key constraints as well :: class User(Base): __tablename__ = 'user' first_name = sa.Column(sa.Unicode(255), primary_key=True) last_name = sa.Column(sa.Unicode(255), primary_key=True) class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) author_first_name = sa.Column(sa.Unicode(255)) author_last_name = sa.Column(sa.Unicode(255)) __table_args__ = ( sa.ForeignKeyConstraint( [author_first_name, author_last_name], [User.first_name, User.last_name] ), sa.Index( 'my_index', author_first_name, author_last_name ) ) table = Article.__table__ constraint = list(table.foreign_keys)[0].constraint has_index(constraint) # True ROnly columns belonging to Table objects are supported. Given column belongs to %r.c3fK|](}t|jj*ywr)r columnsvaluesr&indexrDs rr(zhas_index..s,H9> EMM((*G4Hs.1) tabler,rTable TypeError primary_keyrDrEForeignKeyConstraintlistr anyindexescolumn_or_constraintrH primary_keysrDs @r has_indexrSs| ! & &E eRXX & $&+ ,  $$,,335L&(?(?@+33::<='(  ?[w? CHBG--HErc|j}t|tjst d|zt |j jj}t|tjr$t |jjn|g|k(xs>tfd|jDxstfd|jDS)au Return whether or not given column or given foreign key constraint has a unique index. A column has a unique index if it has a single column primary key index or it has a single column UniqueConstraint. A foreign key constraint has a unique index if the columns of the constraint are the same as the columns of table primary key or the coluns of any unique index or any unique constraint of the given table. :param column: SQLAlchemy Column object .. versionadded: 0.27.1 .. versionchanged: 0.30.18 Added support for foreign key constaints. Fixed support for unique indexes (previously only worked for unique constraints) :: from sqlalchemy_utils import has_unique_index class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) title = sa.Column(sa.String(100)) is_published = sa.Column(sa.Boolean, unique=True) is_deleted = sa.Column(sa.Boolean) is_archived = sa.Column(sa.Boolean) table = Article.__table__ has_unique_index(table.c.is_published) # True has_unique_index(table.c.is_deleted) # False has_unique_index(table.c.id) # True This function supports foreign key constraints as well :: class User(Base): __tablename__ = 'user' first_name = sa.Column(sa.Unicode(255), primary_key=True) last_name = sa.Column(sa.Unicode(255), primary_key=True) class Article(Base): __tablename__ = 'article' id = sa.Column(sa.Integer, primary_key=True) author_first_name = sa.Column(sa.Unicode(255)) author_last_name = sa.Column(sa.Unicode(255)) __table_args__ = ( sa.ForeignKeyConstraint( [author_first_name, author_last_name], [User.first_name, User.last_name] ), sa.Index( 'my_index', author_first_name, author_last_name, unique=True ) ) table = Article.__table__ constraint = list(table.foreign_keys)[0].constraint has_unique_index(constraint) # True :raises TypeError: if given column does not belong to a Table object rBc3K|]X}t|tjjjr(t |j jk(Zywr)r,rsqlschemaUniqueConstraintrMrDrE)r& constraintrDs rr(z#has_unique_index..~sJ *bffmm&D&DE tJ..5578 8 sAA!c3K|]6}|jr(t|jjk(8ywr)uniquerMrDrErFs rr(z#has_unique_index..s7 || tEMM0023 3 s>( #rwwx'@3'F h  #2;0 00s A11A:c t|}|j}|jj}d} |dk(rd|z}|ddddfD]c}t ||}t j |d } tt|t j|c|r|jSS |r|jy y |d k(rbt |d}t j |}d |z}tt|t j||r|jSS|d k(r_t |d}t j |}|r&|dk(xs t| |r|jSS |r|jyy|dk(red|z}t |d}t j |d } tt|t j||r|jSSd} t j |}tt|t j||r|jSS#ttf$rYwxYw#ttf$rY|r|jy y wxYw#ttf$rY|r|jy y wxYw#|r|jwwxYw)aCheck if a database exists. :param url: A SQLAlchemy engine URL. Performs backend-specific testing to quickly determine if a database exists on the server. :: database_exists('postgresql://postgres@localhost/name') #=> False create_database('postgresql://postgres@localhost/name') database_exists('postgresql://postgres@localhost/name') #=> True Supports checking against a constructed URL as well. :: engine = create_engine('postgresql://postgres@localhost/name') database_exists(engine.url) #=> False create_database(engine.url) database_exists(engine.url) #=> True N postgresqlz,SELECT 1 FROM pg_database WHERE datname='%s'postgres template1 template0rk AUTOCOMMITisolation_levelFmysqlzLSELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME = '%s'sqlite:memory:Tmssqlz-SELECT 1 FROM sys.databases WHERE name = '%s'masterzSELECT 1)rrl get_dialectnameror create_engineboolrurdisposerrr)rhrl dialect_namersrdbs rdatabase_existsrs* 3-C||H??$))L F/ < 'AHLD[+tL 'b9))#|L 262774= IJJN  NN  W H  NN  EW $#C$7C%%c*F+-56 *62774=AB6  NN  3X %#C$7C%%c*F:-N1DX1NN*  NN  #"  NN  !W $BXMD#C(;C%%c<HF .vrwwt}EF  NN  D ))#..vrwwt}EF  NN  M)*:;8%&67   NN   %&67   NN     NN  s8J/.'I J/+J/AJ/(:J/ .J/;'I6J/9t|}|j}|jj}|jj}|dk(rt |d}n8|dk(rt |d}n%|dk(rt |d}n|dk(s t |d }|dk(r|d vs |dk(r|d vrt j|d }nt j|}|dk(rj|sd}|j5}djt|||t||}|jt j|d d d n|dk(rZ|j5}djt|||}|jt j|d d d n|dk(ri|dk7rd|r|j5}|jt jd|jt jdd d d nL|j5}dt||}|jt j|d d d |jy #1swYxYw#1swY&xYw#1swY2xYw#1swY>xYw)aIssue the appropriate CREATE DATABASE statement. :param url: A SQLAlchemy engine URL. :param encoding: The encoding to create the database as. :param template: The name of the template from which to create the new database. At the moment only supported by PostgreSQL driver. To create a database, you can pass a simple URL that would have been passed to ``create_engine``. :: create_database('postgresql://postgres@localhost/name') You may also pass the url from an existing engine. :: create_database(engine.url) Has full support for mysql, postgres, and sqlite. In theory, other database engines should be supported. rrrkrr cockroachdb defaultdbrN>pyodbcpymssql>pg8000asyncpgpsycopgpsycopg2 psycopg2cffirrrz,CREATE DATABASE {} ENCODING '{}' TEMPLATE {}rz'CREATE DATABASE {} CHARACTER SET = '{}'rzCREATE TABLE DB(id int)z DROP TABLE DBzCREATE DATABASE )rrlrrdriverrorrbeginformatr executerr) rhencodingtemplaterlrdialect_driverrsrtrs rcreate_databasers], 3-C||H??$))L__&--N|#j9  h7  &k: X %d3N6K$K $  G H!!#|D!!#&|#"H \\^ (tAHHdH%xtX1FD LL '  ( (  \\^ (t<CCdH%xD LL '  ( (  !h*&<  74 RWW%>?@ RWW_56 7 7 \\^ (t%eD(&;%<=D LL ' ( NN1 ( ( ( ( 7 7  ( (s3/A I/AI;A J"4J/I8;JJJct|}|j}|jj}|jj}|dk(rt |d}n8|dk(rt |d}n%|dk(rt |d}n|dk(s t |d }|dk(r|d vs |dk(r|d vrt j|d }nt j|}|dk(r|dk7r|rbtj|nL|j5}dt||}|jt j|d d d |jy #1swYxYw)a>Issue the appropriate DROP DATABASE statement. :param url: A SQLAlchemy engine URL. Works similar to the :func:`create_database` method in that both url text and a constructed url are accepted. :: drop_database('postgresql://postgres@localhost/name') drop_database(engine.url) rrrkrrrrrN>rr>rrrrrrrrzDROP DATABASE )rrlrrrrorrrzremoverr rrr)rhrlrrrsrtrs r drop_databaserTsD 3-C||H??$))L__&--N|#j9  h7  &k: X %d3N6K$K $  G H!!#|D!!#&xH $:  IIh  \\^ (t#E$$9#:;D LL ' ( NN  ( (s 4EE)*)T)utf8N) r.rzcollections.abcrrr sqlalchemyrsqlalchemy.engine.urlrsqlalchemy.excrrutilsr ormr rr$r=rSr]rgrsrhURLrorurrrrrrrs} -*=0BJDNl^gT"299==,,$ 1HVIX.r