引擎是sqlalchemy的核心,不管是 sql core 還是orm的使用都需要依賴引擎的創建,為此我們研究下,引擎是如何創建的。
1 from sqlalchemy import create_engine 2 engine = create_engine('mysql+pymysql://root:x@127.0.0.1/test', 3 echo=True, # 設置為True,則輸出sql語句 4 pool_size=5, # 數據庫連接池初始化的容量 5 max_overflow=10, # 連接池最大溢出容量,該容量+初始容量=最大容量。超出會堵塞等待,等待時間為timeout參數值默認30 6 7 pool_recycle=7200 # 重連周期 8 )
create_engine 創建引擎對象,源代碼如下:
class PlainEngineStrategy(DefaultEngineStrategy): """Strategy for configuring a regular Engine.""" name = "plain" engine_cls = base.Engine PlainEngineStrategy()
這里有個參數 strategy:策略,一般情況默認是'plain',通過參數動態去實例策略類。我們看看對應默認的策略'plain'對應的類是哪個?
default_strategy = "plain" def create_engine(*args, **kwargs): strategy = kwargs.pop("strategy", default_strategy) strategy = strategies.strategies[strategy] return strategy.create(*args, **kwargs)
可以看到是PlainEngineStrategy(),接下來回到創建方法 strategy.create(*args, **kwargs),具體看看怎么創建的。
其實調用了父類DefaultEngineStrategy的方法create。
class DefaultEngineStrategy(EngineStrategy): """Base class for built-in strategies.""" def create(self, name_or_url, **kwargs): # create url.URL object u = url.make_url(name_or_url) plugins = u._instantiate_plugins(kwargs) u.query.pop("plugin", None) kwargs.pop("plugins", None) entrypoint = u._get_entrypoint() dialect_cls = entrypoint.get_dialect_cls(u) if kwargs.pop("_coerce_config", False): def pop_kwarg(key, default=None): value = kwargs.pop(key, default) if key in dialect_cls.engine_config_types: value = dialect_cls.engine_config_types[key](value) return value else: pop_kwarg = kwargs.pop dialect_args = {} # consume dialect arguments from kwargs for k in util.get_cls_kwargs(dialect_cls): if k in kwargs: dialect_args[k] = pop_kwarg(k) dbapi = kwargs.pop("module", None) if dbapi is None: dbapi_args = {} for k in util.get_func_kwargs(dialect_cls.dbapi): if k in kwargs: dbapi_args[k] = pop_kwarg(k) dbapi = dialect_cls.dbapi(**dbapi_args) dialect_args["dbapi"] = dbapi for plugin in plugins: plugin.handle_dialect_kwargs(dialect_cls, dialect_args) # create dialect dialect = dialect_cls(**dialect_args) # assemble connection arguments (cargs, cparams) = dialect.create_connect_args(u) cparams.update(pop_kwarg("connect_args", {})) cargs = list(cargs) # allow mutability # look for existing pool or create pool = pop_kwarg("pool", None) if pool is None: def connect(connection_record=None): if dialect._has_events: for fn in dialect.dispatch.do_connect: connection = fn( dialect, connection_record, cargs, cparams ) if connection is not None: return connection return dialect.connect(*cargs, **cparams) creator = pop_kwarg("creator", connect) poolclass = pop_kwarg("poolclass", None) if poolclass is None: poolclass = dialect_cls.get_pool_class(u) pool_args = {"dialect": dialect} # consume pool arguments from kwargs, translating a few of # the arguments translate = { "logging_name": "pool_logging_name", "echo": "echo_pool", "timeout": "pool_timeout", "recycle": "pool_recycle", "events": "pool_events", "use_threadlocal": "pool_threadlocal", "reset_on_return": "pool_reset_on_return", "pre_ping": "pool_pre_ping", "use_lifo": "pool_use_lifo", } for k in util.get_cls_kwargs(poolclass): tk = translate.get(k, k) if tk in kwargs: pool_args[k] = pop_kwarg(tk) for plugin in plugins: plugin.handle_pool_kwargs(poolclass, pool_args) pool = poolclass(creator, **pool_args) else: if isinstance(pool, poollib.dbapi_proxy._DBProxy): pool = pool.get_pool(*cargs, **cparams) else: pool = pool pool._dialect = dialect # create engine. engineclass = self.engine_cls engine_args = {} for k in util.get_cls_kwargs(engineclass): if k in kwargs: engine_args[k] = pop_kwarg(k) _initialize = kwargs.pop("_initialize", True) # all kwargs should be consumed if kwargs: raise TypeError( "Invalid argument(s) %s sent to create_engine(), " "using configuration %s/%s/%s. Please check that the " "keyword arguments are appropriate for this combination " "of components." % ( ",".join("'%s'" % k for k in kwargs), dialect.__class__.__name__, pool.__class__.__name__, engineclass.__name__, ) ) engine = engineclass(pool, dialect, u, **engine_args) if _initialize: do_on_connect = dialect.on_connect() if do_on_connect: def on_connect(dbapi_connection, connection_record): conn = getattr( dbapi_connection, "_sqla_unwrap", dbapi_connection ) if conn is None: return do_on_connect(conn) event.listen(pool, "first_connect", on_connect) event.listen(pool, "connect", on_connect) def first_connect(dbapi_connection, connection_record): c = base.Connection( engine, connection=dbapi_connection, _has_events=False ) c._execution_options = util.immutabledict() dialect.initialize(c) dialect.do_rollback(c.connection) event.listen(pool, "first_connect", first_connect, once=True) dialect_cls.engine_created(engine) if entrypoint is not dialect_cls: entrypoint.engine_created(engine) for plugin in plugins: plugin.engine_created(engine) return engine
我們逐一分析:
u = url.make_url(name_or_url) # 這個方法解析傳入的數據庫連接的uri信息,符合條件最終返回一個URL對象
plugins = u._instantiate_plugins(kwargs) # 插件初始化,
entrypoint = u._get_entrypoint() # 根據傳入url中的數據庫類型(mysql)和驅動庫(pymysql),來注冊插件,返回方言類
dialect_cls = entrypoint.get_dialect_cls(u) # 獲取Dialect類
這里需要說明下Dialect(方言類)的作用是用來定義數據庫和DBapi的行為
if kwargs.pop("_coerce_config", False): def pop_kwarg(key, default=None): value = kwargs.pop(key, default) if key in dialect_cls.engine_config_types: value = dialect_cls.engine_config_types[key](value) return value else: pop_kwarg = kwargs.pop dialect_args = {} # consume dialect arguments from kwargs for k in util.get_cls_kwargs(dialect_cls): if k in kwargs: dialect_args[k] = pop_kwarg(k)
這段代碼沒啥,創建出方言所需要的完整參數dialect_args
dbapi = kwargs.pop("module", None) if dbapi is None: dbapi_args = {} for k in util.get_func_kwargs(dialect_cls.dbapi): if k in kwargs: dbapi_args[k] = pop_kwarg(k) dbapi = dialect_cls.dbapi(**dbapi_args) dialect_args["dbapi"] = dbapi
這段代碼 則是實例化dbpai對象。
# create dialect dialect = dialect_cls(**dialect_args)
開始實例化方言
(cargs, cparams) = dialect.create_connect_args(u) cparams.update(pop_kwarg("connect_args", {})) cargs = list(cargs) # allow mutability
創建連接所需要的參數
pool = pop_kwarg("pool", None) if pool is None: def connect(connection_record=None): if dialect._has_events: for fn in dialect.dispatch.do_connect: connection = fn( dialect, connection_record, cargs, cparams ) if connection is not None: return connection return dialect.connect(*cargs, **cparams) creator = pop_kwarg("creator", connect) poolclass = pop_kwarg("poolclass", None) if poolclass is None: poolclass = dialect_cls.get_pool_class(u) pool_args = {"dialect": dialect} # consume pool arguments from kwargs, translating a few of # the arguments translate = { "logging_name": "pool_logging_name", "echo": "echo_pool", "timeout": "pool_timeout", "recycle": "pool_recycle", "events": "pool_events", "use_threadlocal": "pool_threadlocal", "reset_on_return": "pool_reset_on_return", "pre_ping": "pool_pre_ping", "use_lifo": "pool_use_lifo", } for k in util.get_cls_kwargs(poolclass): tk = translate.get(k, k) if tk in kwargs: pool_args[k] = pop_kwarg(tk) for plugin in plugins: plugin.handle_pool_kwargs(poolclass, pool_args) pool = poolclass(creator, **pool_args) else: if isinstance(pool, poollib.dbapi_proxy._DBProxy): pool = pool.get_pool(*cargs, **cparams) else: pool = pool pool._dialect = dialect
創建連接池,默認創建pool.QueuePool
# create engine. engineclass = self.engine_cls engine_args = {} for k in util.get_cls_kwargs(engineclass): if k in kwargs: engine_args[k] = pop_kwarg(k) _initialize = kwargs.pop("_initialize", True) # all kwargs should be consumed if kwargs: raise TypeError( "Invalid argument(s) %s sent to create_engine(), " "using configuration %s/%s/%s. Please check that the " "keyword arguments are appropriate for this combination " "of components." % ( ",".join("'%s'" % k for k in kwargs), dialect.__class__.__name__, pool.__class__.__name__, engineclass.__name__, ) ) engine = engineclass(pool, dialect, u, **engine_args)
從上面可以看出來,引擎的核心是連接池和方言,連接池負責連接的維護,方言負責數據的行為。
if _initialize: do_on_connect = dialect.on_connect() if do_on_connect: def on_connect(dbapi_connection, connection_record): conn = getattr( dbapi_connection, "_sqla_unwrap", dbapi_connection ) if conn is None: return do_on_connect(conn) event.listen(pool, "first_connect", on_connect) event.listen(pool, "connect", on_connect) def first_connect(dbapi_connection, connection_record): c = base.Connection( engine, connection=dbapi_connection, _has_events=False ) c._execution_options = util.immutabledict() dialect.initialize(c) dialect.do_rollback(c.connection) event.listen(pool, "first_connect", first_connect, once=True)
第一次初始化連接並進行監聽
dialect_cls.engine_created(engine) if entrypoint is not dialect_cls: entrypoint.engine_created(engine) for plugin in plugins: plugin.engine_created(engine)
總結:
create_engine通過傳入的URI和相關參數,創建一個Engine,該引擎包含了方言(Dialect)和Pool,Dialect如中文名翻譯一樣,方言:作為不同的數據庫Mysql,Oracle,PostgreSQL等,會有不同的行為,Dialect就是用來操作不同數據庫的行為,對應接口調用dbapi操作。
而Pool作為數據庫連接池,用來管理數據庫連接,通過維護一個連接池,池子的大小,數量和生命周期,減少數據庫連接的頻繁切換,提高查詢等操作效率。