neuralib.sqlp.stat_start

neuralib.sqlp.stat_start.create_table(table, *, if_not_exists=True)[source]

CREATE: https://www.sqlite.org/lang_createtable.html

>>> @named_tuple_table_class 
... class A(NamedTuple):
...     a: int
>>> create_table(A) 
CREATE TABLE IF NOT EXISTS A (a INT NOT NULL)

features supporting

  • IF NOT EXISTS

  • column constraint NOT NULL

  • column constraint PRIMARY KEY

  • column constraint UNIQUE

  • column constraint CHECK

  • column constraint DEFAULT value

  • table constraint PRIMARY KEY

  • table constraint UNIQUE

  • table constraint CHECK

  • table constraint FOREIGN KEY

features not supporting

  • CREATE TEMP|TEMPORARY

  • CREATE TEMP

  • AS SELECT

  • column constraint CONSTRAINT

  • column constraint NOT NULL ON CONFLICT

  • column constraint DEFAULT (EXPR)

  • column constraint COLLATE

  • column constraint REFERENCES

  • column constraint [GENERATED ALWAYS] AS

  • table constraint CONSTRAINT

  • WITHOUT ROWID

  • STRICT

Parameters:

table (type[T])

Returns:

Return type:

SqlStat[T]

neuralib.sqlp.stat_start.insert_into(table: type[T], *, policy: Literal['ABORT', 'FAIL', 'IGNORE', 'REPLACE', 'ROLLBACK'] = None, named=False) SqlInsertStat[T][source]
neuralib.sqlp.stat_start.insert_into(*field, policy: Literal['ABORT', 'FAIL', 'IGNORE', 'REPLACE', 'ROLLBACK'] = None, named=False) SqlInsertStat[T]

INSERT: https://www.sqlite.org/lang_insert.html

insert values

>>> insert_into(A, policy='REPLACE').build() 
INSERT OR REPLACE INTO A VALUES (?)
>>> insert_into(A, policy='REPLACE').submit([A(1), A(2)]) 

insert values with field overwrite

>>> insert_into(A, policy='REPLACE').values(a='1').build() 
INSERT OR REPLACE INTO A VALUES (1)

insert values from a table

>>> insert_into(A, policy='IGNORE').select_from(B).build() 
INSERT OR IGNORE INTO A
SELECT * FROM B

features supporting

  • INSERT [OR …]

  • VALUES

  • DEFAULT VALUES

  • SELECT

  • upsert clause

  • returning clause

features not supporting

  • WITH [RECURSIVE]

Parameters:
  • table

  • policy

  • named

Returns:

neuralib.sqlp.stat_start.replace_into(table: type[T], *, named=False) SqlInsertStat[T][source]
neuralib.sqlp.stat_start.replace_into(*field: Any, named=False) SqlInsertStat[T]
neuralib.sqlp.stat_start.select_from(table: type[T], *, distinct: bool = False) SqlSelectStat[T][source]
neuralib.sqlp.stat_start.select_from(table: SqlCteExpr, *, distinct: bool = False) SqlSelectStat[tuple]
neuralib.sqlp.stat_start.select_from(*field, distinct: bool = False, from_table: str | type | SqlAlias | SqlSelectStat = None) SqlSelectStat[tuple]

SELECT: https://www.sqlite.org/lang_select.html

Select all fields from a table

>>> select_from(A).build() 
SELECT * FROM A
>>> select_from(A).fetchall() 
[A(...), A(...), ...]

Select subset of fields from A

>>> select_from(A.a, A.b).build() 
SELECT A.a, A.b FROM A
>>> select_from(A.a, A.b).fetchall() 
[('a', 1), ('b', 2), ...]

With a literal value

>>> select_from(A.a, 0).build() 
SELECT A.a, 0 FROM A
>>> select_from(A.a, 0).fetchall() 
[('a', 0), ('b', 0), ...]

With SQL functions

>>> select_from(A.a, count()).build() 
SELECT A.a, COUNT(*) FROM A

Use table alias

>>> a = alias(A, 'a') 
>>> select_from(a.a).build() 
SELECT a.a from A a

join other tables

>>> select_from(A.a, B.b).join(A.c == B.c).build() 
SELECT A.a, B.b FROM A JOIN B ON A.c = B.c

features supporting

  • SELECT DISTINCT

  • FROM

  • WHERE

  • GROUP BY

  • HAVING

  • WINDOW

  • compound-operator: UNION [ALL], INTERSECT and EXCEPT

  • ORDER BY

  • LIMIT [OFFSET]

features not supporting

  • WITH [RECURSIVE]

  • SELECT ALL

  • VALUES

Parameters:
  • args

  • distinct

  • from_table

Returns:

neuralib.sqlp.stat_start.update(table, *args, **kwargs)[source]

UPDATE: https://www.sqlite.org/lang_update.html

>>> update(A, A.a==1).where(A.b==2).build() 
UPDATE A SET A.a = 1 WHERE A.b = 2

features supporting

  • UPDATE [OR …]

  • SET COLUMN = EXPR

  • FROM

  • WHERE

  • ON CONFLICT (COLUMNS) SET (COLUMNS) = EXPR

  • returning clause

features not supporting

  • WITH [RECURSIVE]

  • (qualified table name) INDEXED BY

  • (qualified table name) NOT INDEXED

Parameters:
Returns:

Return type:

SqlUpdateStat[T]

neuralib.sqlp.stat_start.delete_from(table)[source]

DELETE: https://www.sqlite.org/lang_delete.html

>>> delete_from(A).where(A.b > 2).build()  
DELETE FROM A WHERE A.b > 2

features supporting

  • DELETE FROM

  • WHERE

  • ORDER BY

  • LIMIT [OFFSET]

  • returning clause

features not supporting

  • WITH [RECURSIVE]

  • (qualified table name) INDEXED BY

  • (qualified table name) NOT INDEXED

Parameters:

table (type[T])

Returns:

Return type:

SqlDeleteStat[T]