import json

import sqlite3

from typing import Any, Callable, Dict, List, Optional, Type, Union

from sqlalchemy import create_engine as _create_engine

from sqlalchemy.engine.url import URL

from sqlalchemy.future import Engine as _FutureEngine

from sqlalchemy.pool import Pool

from typing_extensions import Literal, TypedDict

from ..default import Default, _DefaultPlaceholder

# Types defined in sqlalchemy2-stubs, but can't be imported, so re-define here

_Debug = Literal["debug"]

_IsolationLevel = Literal[

"SERIALIZABLE",

"REPEATABLE READ",

"READ COMMITTED",

"READ UNCOMMITTED",

"AUTOCOMMIT",

]

_ParamStyle = Literal["qmark", "numeric", "named", "format", "pyformat"]

_ResetOnReturn = Literal["rollback", "commit"]

class _SQLiteConnectArgs(TypedDict, total=False):

timeout: float

detect_types: Any

isolation_level: Optional[Literal["DEFERRED", "IMMEDIATE", "EXCLUSIVE"]]

check_same_thread: bool

factory: Type[sqlite3.Connection]

cached_statements: int

uri: bool

_ConnectArgs = Union[_SQLiteConnectArgs, Dict[str, Any]]

# Re-define create_engine to have by default future=True, and assume that's what is used

# Also show the default values used for each parameter, but don't set them unless

# explicitly passed as arguments by the user to prevent errors. E.g. SQLite doesn't

# support pool connection arguments.

def create_engine(

url: Union[str, URL],

*,

connect_args: _ConnectArgs = Default({}), # type: ignore

echo: Union[bool, _Debug] = Default(False),

echo_pool: Union[bool, _Debug] = Default(False),

enable_from_linting: bool = Default(True),

encoding: str = Default("utf-8"),

execution_options: Dict[Any, Any] = Default({}),

future: bool = True,

hide_parameters: bool = Default(False),

implicit_returning: bool = Default(True),

isolation_level: Optional[_IsolationLevel] = Default(None),

json_deserializer: Callable[..., Any] = Default(json.loads),

json_serializer: Callable[..., Any] = Default(json.dumps),

label_length: Optional[int] = Default(None),

logging_name: Optional[str] = Default(None),

max_identifier_length: Optional[int] = Default(None),

max_overflow: int = Default(10),

module: Optional[Any] = Default(None),

paramstyle: Optional[_ParamStyle] = Default(None),

pool: Optional[Pool] = Default(None),

poolclass: Optional[Type[Pool]] = Default(None),

pool_logging_name: Optional[str] = Default(None),

pool_pre_ping: bool = Default(False),

pool_size: int = Default(5),

pool_recycle: int = Default(-1),

pool_reset_on_return: Optional[_ResetOnReturn] = Default("rollback"),

pool_timeout: float = Default(30),

pool_use_lifo: bool = Default(False),

plugins: Optional[List[str]] = Default(None),

query_cache_size: Optional[int] = Default(None),

**kwargs: Any,

) -> _FutureEngine:

current_kwargs: Dict[str, Any] = {

"future": future,

}

if not isinstance(echo, _DefaultPlaceholder):

current_kwargs["echo"] = echo

if not isinstance(echo_pool, _DefaultPlaceholder):

current_kwargs["echo_pool"] = echo_pool

if not isinstance(enable_from_linting, _DefaultPlaceholder):

current_kwargs["enable_from_linting"] = enable_from_linting

if not isinstance(connect_args, _DefaultPlaceholder):

current_kwargs["connect_args"] = connect_args

if not isinstance(encoding, _DefaultPlaceholder):

current_kwargs["encoding"] = encoding

if not isinstance(execution_options, _DefaultPlaceholder):

current_kwargs["execution_options"] = execution_options

if not isinstance(hide_parameters, _DefaultPlaceholder):

current_kwargs["hide_parameters"] = hide_parameters

if not isinstance(implicit_returning, _DefaultPlaceholder):

current_kwargs["implicit_returning"] = implicit_returning

if not isinstance(isolation_level, _DefaultPlaceholder):

current_kwargs["isolation_level"] = isolation_level

if not isinstance(json_deserializer, _DefaultPlaceholder):

current_kwargs["json_deserializer"] = json_deserializer

if not isinstance(json_serializer, _DefaultPlaceholder):

current_kwargs["json_serializer"] = json_serializer

if not isinstance(label_length, _DefaultPlaceholder):

current_kwargs["label_length"] = label_length

if not isinstance(logging_name, _DefaultPlaceholder):

current_kwargs["logging_name"] = logging_name

if not isinstance(max_identifier_length, _DefaultPlaceholder):

current_kwargs["max_identifier_length"] = max_identifier_length

if not isinstance(max_overflow, _DefaultPlaceholder):

current_kwargs["max_overflow"] = max_overflow

if not isinstance(module, _DefaultPlaceholder):

current_kwargs["module"] = module

if not isinstance(paramstyle, _DefaultPlaceholder):

current_kwargs["paramstyle"] = paramstyle

if not isinstance(pool, _DefaultPlaceholder):

current_kwargs["pool"] = pool

if not isinstance(poolclass, _DefaultPlaceholder):

current_kwargs["poolclass"] = poolclass

if not isinstance(pool_logging_name, _DefaultPlaceholder):

current_kwargs["pool_logging_name"] = pool_logging_name

if not isinstance(pool_pre_ping, _DefaultPlaceholder):

current_kwargs["pool_pre_ping"] = pool_pre_ping

if not isinstance(pool_size, _DefaultPlaceholder):

current_kwargs["pool_size"] = pool_size

if not isinstance(pool_recycle, _DefaultPlaceholder):

current_kwargs["pool_recycle"] = pool_recycle

if not isinstance(pool_reset_on_return, _DefaultPlaceholder):

current_kwargs["pool_reset_on_return"] = pool_reset_on_return

if not isinstance(pool_timeout, _DefaultPlaceholder):

current_kwargs["pool_timeout"] = pool_timeout

if not isinstance(pool_use_lifo, _DefaultPlaceholder):

current_kwargs["pool_use_lifo"] = pool_use_lifo

if not isinstance(plugins, _DefaultPlaceholder):

current_kwargs["plugins"] = plugins

if not isinstance(query_cache_size, _DefaultPlaceholder):

current_kwargs["query_cache_size"] = query_cache_size

current_kwargs.update(kwargs)

return _create_engine(url, **current_kwargs)