Skip to content

Commit

Permalink
Fix linting, typing
Browse files Browse the repository at this point in the history
  • Loading branch information
karpetrosyan committed Sep 30, 2023
1 parent 11214d1 commit f11774e
Show file tree
Hide file tree
Showing 2 changed files with 37 additions and 10 deletions.
21 changes: 21 additions & 0 deletions anysqlite/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,4 +19,25 @@

from ._core import Connection, Cursor, connect # noqa: F401

__all__ = [
"Connection",
"Cursor",
"connect",
"DatabaseError",
"DataError",
"Error",
"IntegrityError",
"InterfaceError",
"InternalError",
"NotSupportedError",
"OperationalError",
"ProgrammingError",
"Row",
"Warning",
"apilevel",
"paramstyle",
"sqlite_version",
"sqlite_version_info",
"threadsafety",
]
__version__ = "0.0.3"
26 changes: 16 additions & 10 deletions anysqlite/_core.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
import sqlite3
import typing as tp
from functools import partial, wraps
from functools import partial
from pathlib import Path

from anyio import CapacityLimiter, to_thread
from pathlib import Path


class Connection:
def __init__(self, _real_connection: sqlite3.Connection) -> None:
Expand Down Expand Up @@ -38,7 +39,11 @@ def __init__(self, real_cursor: sqlite3.Cursor, limiter: CapacityLimiter) -> Non
self._limiter = limiter

@property
def description(self) -> tp.Union[tp.Tuple[tp.Tuple[str, None, None, None, None, None, None], ...], tp.Any]:
def description(
self,
) -> tp.Union[
tp.Tuple[tp.Tuple[str, None, None, None, None, None, None], ...], tp.Any
]:
return self._real_cursor.description

@property
Expand All @@ -50,17 +55,17 @@ def arraysize(self) -> int:
return self._real_cursor.arraysize

async def close(self) -> None:
await to_thread.run_sync(
self._real_cursor.close, limiter=self._limiter
)
await to_thread.run_sync(self._real_cursor.close, limiter=self._limiter)

async def execute(self, sql: str, parameters: tp.Iterable[tp.Any] = ()) -> "Cursor":
real_cursor = await to_thread.run_sync(
self._real_cursor.execute, sql, parameters, limiter=self._limiter
)
return Cursor(real_cursor, self._limiter)

async def executemany(self, sql: str, seq_of_parameters: tp.Iterable[tp.Iterable[tp.Any]]) -> "Cursor":
async def executemany(
self, sql: str, seq_of_parameters: tp.Iterable[tp.Iterable[tp.Any]]
) -> "Cursor":
real_cursor = await to_thread.run_sync(
self._real_cursor.executemany, sql, seq_of_parameters, limiter=self._limiter
)
Expand All @@ -87,9 +92,10 @@ async def fetchall(self) -> tp.Any:
self._real_cursor.fetchall, limiter=self._limiter
)

async def connect(database: tp.Union[str, bytes, Path],
**kwargs: tp.Any) -> "Connection":


async def connect(
database: tp.Union[str, bytes, Path], **kwargs: tp.Any
) -> "Connection":
kwargs["check_same_thread"] = False
real_connection = await to_thread.run_sync(
partial(sqlite3.connect, database, **kwargs)
Expand Down

0 comments on commit f11774e

Please sign in to comment.