sqlmodel/sqlmodel/engine/create.py at main · mrdeveloperdude/sqlmodel · GitHub

Repository navigation

History

139 lines (128 loc) · 6.08 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

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)

sqlmodel/sqlmodel/engine/create.py at main · mrdeveloperdude/sqlmodel · GitHub

Repository navigation

Expand file tree

Breadcrumbs

create.py

Latest commit

History

Breadcrumbs

create.py

File metadata and controls