neuralib.sqlp.stat

class neuralib.sqlp.stat.SqlStat[source]

Abstract SQL statement.

__init__(table)[source]
Parameters:

table (type[T] | None)

build()[source]

build a SQL statement.

Return type:

tuple[str, list]

submit()[source]

build the SQL statement and execute.

Returns:

a cursor

Raises:

RuntimeError – current statement does not bind with a connection.

Return type:

Cursor[T]

drop()[source]
add(stat)[source]

Add SQL token.

Return type:

Self

class neuralib.sqlp.stat.SqlSelectStat[source]

SELECT statement.

__init__(table)[source]
Parameters:

table (type[T] | None)

fetchall()[source]

submit and fetch all result.

Return type:

list[T]

fetchone()[source]

submit and fetch the first result.

Return type:

T | None

fetch_polars()[source]

submit and fetch all as a polar dataframe.

Return type:

DataFrame

windows(*windows, **window_ks)[source]

define windows.

Parameters:
Return type:

Self

BY

alias of Literal[‘left’, ‘right’, ‘inner’, ‘full outer’, ‘cross’]

join(constraint: Callable | ForeignConstraint, *, by: BY = None) → SqlSelectStat[tuple][source]
join(table: type[S] | SqlAlias[S], constraint: Callable | ForeignConstraint, *, by: BY = None) → SqlSelectStat[tuple]
join(table: type[S] | SqlSelectStat[S] | SqlAlias[S] | SqlCteExpr, *field: bool | Any, by: BY = None) → SqlSelectStat[tuple]
join(*field: bool | Any, by: BY = None) → SqlSelectStat[tuple]

JOIN https://www.sqlite.org/lang_select.html#strange_join_names

>>> select_from(A.a, B.b).join(A.a == B.a) 
SELECT A.a, B.b FROM A
JOIN B ON A.a = B.a
group_by(*by)[source]

GROUP BY https://www.sqlite.org/lang_select.html#resultset

Return type:

Self

having(*exprs)[source]

HAVING https://www.sqlite.org/lang_select.html#resultset

Parameters:

exprs (bool | SqlExpr)

Return type:

Self

intersect(stat)[source]

INTERSECT https://www.sqlite.org/lang_select.html#compound_select_statements

Parameters:

stat (SqlStat)

Return type:

Self

union(stat, all=False)[source]

UNION https://www.sqlite.org/lang_select.html#compound_select_statements

Parameters:

stat (SqlStat)

Return type:

Self

except_(stat)[source]

EXCEPT https://www.sqlite.org/lang_select.html#compound_select_statements

Parameters:

stat (SqlStat)

Return type:

Self

class neuralib.sqlp.stat.SqlInsertStat[source]
__init__(table, fields=None, *, named=False)[source]
Parameters:

  • table (type[T])

  • fields (list[str] | None)

  • named (bool)

select_from(table: type[T], *, distinct: bool = False) → SqlSelectStat[T][source]
select_from(*field, distinct: bool = False, from_table: str | type | SqlAlias | SqlSelectStat = None) → SqlSelectStat[tuple]
values(*args, **kwargs)[source]
Parameters:

kwargs (str | SqlExpr)

Return type:

Self

defaults()[source]

insert default values

Return type:

Self

on_conflict(*conflict, where=None)[source]
Parameters:

where (bool | SqlCompareOper)

Return type:

SqlUpsertStat[T]

returning(*expr)[source]
Parameters:

expr (str | SqlExpr)

Return type:

SqlInsertStat[tuple]

build()[source]

build a SQL statement.

Return type:

tuple[str, list]

submit(parameter=())[source]

build the SQL statement and execute.

Returns:

a cursor

Raises:

RuntimeError – current statement does not bind with a connection.

Parameters:

parameter (list[T])

Return type:

Cursor[T]

class neuralib.sqlp.stat.SqlUpdateStat[source]
from_(query)[source]
Parameters:

query (SqlStat | SqlAlias[SqlSubQuery])

Return type:

Self

class neuralib.sqlp.stat.SqlDeleteStat[source]
class neuralib.sqlp.stat.Cursor[source]

A SQL cursor wrapper.

It will try to cast to T from the tuple returns.

__init__(connection, cursor, table=None)[source]
Parameters:

  • connection (Connection)

  • cursor (sqlite3.Cursor)

  • table (type[T])

property headers: list[str]
fetchall()[source]

fetch all results.

Return type:

list[T]

fetchone()[source]

fetch the first result.

Return type:

T | None

fetch_polars()[source]
Return type:

DataFrame