L i('dZddlmZddlZddlZddlZddlmZmZddl m Z m Z ddl m Z mZmZddlmZddlmZGd d Zy) z"SQLite abstraction for coverage.py) annotationsN)IterableIterator)Anycast) auto_repr clipped_repr exc_one_line) DataError) TDebugCtlceZdZdZdddZeZddZdddZddZ ddZ ddZ e j d ddZddd Zddd Zddd Zdd Zdd ZddZd dZy)!SqliteDbaUA simple abstraction over a SQLite database. Use as a context manager, then you can use it like a :class:`python:sqlite3.Connection` object:: with SqliteDb(filename, debug_control) as db: with db.execute("select a, b from some_table") as cur: for a, b in cur: etc(a, b) cJ||_||_||_d|_d|_y)Nr)debugfilenameno_disknestcon)selfrrrs W/mnt/ssd/data/python-lab/Trading/venv/lib/python3.12/site-packages/coverage/sqlitedb.py__init__zSqliteDb.__init__ s%     .2c~|jy|jjdr(|jjd|j |jj dr(t j|jdd|_n&t j|jd|_|jjdr5|jjd |jd |j|jjd ddtt dr*|jjt jd|jd|jddy#t j$r!}td |jd ||d}~wwxYw)z2Connect to the db and do universal initialization.NsqlzConnecting to zfile:FT)check_same_threaduri)rCouldn't use data file : z Connected to z as REGEXPc2tj||duS)N)research)txtpats rz#SqliteDb._connect..Asryyc?RZ^?^rSQLITE_DBCONFIG_DEFENSIVEzpragma journal_mode=offzpragma synchronous=off)fail_ok)rrshouldwriter startswithsqlite3connectErrorr create_functionhasattr setconfigr' execute_void)rexcs r_connectzSqliteDb._connect)sf 88   ::  U # JJ  ~dmm->? @ Y}}''0"??4==EW[\"??4==ER ::  U # JJ  }T]],=T$((N O   1.^_ 77 8 HH  11  34 2DA1}} Y5dmm5FbNOUX X YsA)FF<F77F<c|j|s |jsr|jjdr5|jj d|jd|j |jj d|_yyy)z If needed, close the connection.NrzClosing z on )rrrr)r*rclose)rforces rr6zSqliteDb.closeUsn 88 DLL::$$U+JJ$$x|4 ?P%QR  ) rc|jdk(r8|j|jJ|jj|xjdz c_|S)Nr)rr4r __enter__rs rr:zSqliteDb.__enter__^sH 99> MMO88' '' HH    Q  rc|xjdzc_|jdk(r= |jJ|jj||||jyy#t$rc}|j j dr'|j jdt|td|jd||d}~wwxYw)Nr9rrzEXCEPTION from __exit__: zCouldn't end data file r) rr__exit__r6 Exceptionrr)r*r r r)rexc_type exc_value tracebackr3s rr=zSqliteDb.__exit__fs Q 99> ]xx+++!!(IyA   ]::$$U+JJ$$'@cAR@S%TU"9$--9J"SE RSY\\ ]s;A## C,AC  Cc|jjdr)|rd|nd}|jjd|| |jJ |jj ||S#t $r|jj ||cYSwxYw#t j$r}t|}|js^ t|jd5}d}|jt||k(rd}dddn #1swYnxYwn#t $rYnwxYw|jjdr'|jjd t|td |jd ||d}~wwxYw) z2Same as :meth:`python:sqlite3.Connection.execute`.r with z Executing Nrbs&!coverage.py: This is a private formatzILooks like a coverage 4.x data file. Are you mixing versions of coverage?zEXCEPTION from execute: rr)rr)r*rexecuter>r,r.strropenrreadlenr r )rr parameterstailr3msgbad_filecov4_sigs r_executezSqliteDb._executersy ::  U #.8VJ>*bD JJ  z#7 8 Y88' '' 9xx''Z88 9xx''Z88  9 }} Yc(C<< dmmT2h#L#==X78C!I !zz  '   #;L Y!^c3ha1 1 4w.@T 5!QR R  s A;;Bc|jjdr|jjdrdnd}|jjd|dt|d||jjdr4t |D]&\}}|jj|dd |(|j J |j j ||S#t$r|j j ||cYSwxYw) z6Same as :meth:`python:sqlite3.Connection.executemany`.rrU:rDzExecuting many rCrZ4dr)rr)r*rJ enumerater executemanyr>)rrdatafinalirows r _executemanyzSqliteDb._executemanys ::  U #::,,Y7CRE JJ  sgVCI;eE7S Tzz  +'o9FAsJJ$$"vRw%789xx### 388''T2 2 388''T2 2  3sC""%D  D cD|j||jy)zUSame as :meth:`python:sqlite3.Connection.executemany` when you don't need the cursor.N)rir6)rrres rexecutemany_voidzSqliteDb.executemany_voids #t$**,rc $|jjdr>|jjdjt |t |d|j J|j j|jy)z8Same as :meth:`python:sqlite3.Connection.executescript`.rz"Executing script with {} chars: {}dN) rr)r*formatrJr r executescriptr6)rscripts rrozSqliteDb.executescriptsr ::  U # JJ  4;;K -  xx### v&,,.rcp|jJdj|jjS)z9Return a multi-line string, the SQL dump of the database. )rjoiniterdumpr;s rdumpz SqliteDb.dumps.xx###yy**,--rN)F)rrGrr rboolreturnNone)rwrx)r7rvrwrx)rwr)rrGrK Iterable[Any]rwsqlite3.Cursor))rrGrKryrwzIterator[sqlite3.Cursor])r{F)rrGrKryr(rvrwrx)rrGrKryrwint)rrGrKryrwztuple[Any, ...] | None)rrGre list[Any]rwrz)rrGrer}rwrx)rprGrwrx)rwrG)__name__ __module__ __qualname____doc__rr__repr__r4r6r:r=rP contextlibcontextmanagerrFr2rXr_rirkrorur{rrrrs 3H*BX  ]YB%'   "  "   S$3"- /.rr)r __future__rrr"r,collections.abcrrtypingrrcoverage.debugrr r coverage.exceptionsr coverage.typesr rr{rrrs4)" .@@)$\.\.r