SQLAlchemy is the Python SQL toolkit and ORM framework that provides application developers with comprehensive and flexible SQL capabilities. It provides a suite of enterprise-level persistence solutions designed for efficient, high-performance database access and Pythonic zen. The project code is quite large, nearly 200 files, 70,000 lines of code, let’s challenge together. Due to the length, this part is divided into two parts. The first part is about engine, Dialect, Connection and pool of core. The second part is about SQL expressions and ORM of core, including the following contents:
- Example of sqL-schema
- Data Definition Language (DDL) Creates a table
- Data Manipulation Language (DML) uses INSERTS to insert Data
- Data Query Language (DQL) uses SELECT to Query Data
- The ORM example
- Core Functions of Model
- summary
- tip
Example of sqL-schema
In the previous article, the SQL we used was all hand-written statements like this:
create table x (a integer, b integer)
insert into x (a, b) values (1, 1)
Copy the codeSqlalchemy allows data manipulation by defining a schema, as shown in the following complete example:
from sqlalchemy import create_engine from sqlalchemy import MetaData from sqlalchemy import Table from sqlalchemy import Column from sqlalchemy import Integer from sqlalchemy import String from sqlalchemy.sql import select engine = create_engine('sqlite:///:memory:', echo=True) metadata = MetaData() users = Table('users', metadata, Column('id', Integer, primary_key=True), Column('name', String), Column('fullname', String), ) metadata.create_all(engine) ins = users.insert().values(name='jack', fullname='Jack Jones') print(ins) result = engine.execute(ins) print(result, result.inserted_primary_key) s = select([users]) result = conn.execute(s) for row in result: print(row) result = engine.execute("select * from users") for row in result: print(row)Copy the codeThe execution of the sample program:
- Create engine for database connection
- Create metadata to manage schemas
- Create the users Table and bind it to the metadata. Include the id, name, and fullname columns
- Submit metadata to Engine (create tables)
- Insert data using Users
- Query users’ data
- Validate data using plain SQL
Here is the execution log for the example, which clearly shows the above process:
. The 2021-04-19 10:02:09, 166 INFO sqlalchemy. Engine. Base. The engine the CREATE TABLE users (id INTEGER NOT NULL, the name VARCHAR, fullname VARCHAR, PRIMARY KEY (id)) the 2021-04-19 10:02:09, 166 INFO. Sqlalchemy engine. Base. The engine () the 2021-04-19 10:02:09, 167 INFO sqlalchemy.engine.base.Engine COMMIT INSERT INTO users (name, fullname) VALUES (:name, : the 2021-04-19 10:02:09 fullname), 167 INFO sqlalchemy. Engine. Base. The engine INSERT INTO users (name, fullname) VALUES (?,?,?,?,? ,?) The 2021-04-19 10:02:09, 168 INFO sqlalchemy. Engine. The base engine (' jack ', 'Jack Jones') the 2021-04-19 10:02:09, 168 INFO. Sqlalchemy engine. Base. The engine COMMIT < sqlalchemy. Engine. The result. ResultProxy Object at 0 x7ffca0607070 > [1] the 2021-04-27 11:38:19, 134 INFO sqlalchemy. Engine. Base. The engine SELECT users. The id of the users. The name, Users. The fullname FROM users 11:38:19 2021-04-27, 134: INFO sqlalchemy. Engine. Base. The engine () (1, 'jack', 'Jack Jones') the 2021-04-19 10:02:09, 168 INFO. Sqlalchemy engine. Base. The engine a select * from users 10:02:09 2021-04-19, 168 INFO sqlalchemy.engine.base.Engine () (1, 'jack', 'Jack Jones')Copy the codeBefore we begin, we need to take a quick look at the classification of SQL statements:
In our schema usage example, there are three types of statements: DDL, DML, and DQL. Let’s take a look at the SQL expressions section of SQLAlchemy based on these three types. SQL expressions are mainly stored in SQL packages. Some files provide the following functions:
| The module | describe |
|---|---|
| base.py | The base class |
| compiler.py | The SQL compiler |
| crud.py | Crud parameter processing |
| ddl.py | DDL statements |
| default_comparator.py | To compare |
| dml.py | DML statements |
| elements.py | Basic types of |
| operators.py | SQL operator |
| schema.py | Schema definition |
| selectable.py | DQL |
| sqltypes.py&&type_api.py | SQL data types |
| vistitors.py | A recursive algorithm |
Data Definition Language (DDL) Creates a table
First, look at the basic implementation of the Schema visitable:
class VisitableType(type): def __init__(cls, clsname, bases, clsdict): if clsname ! = "Visitable" and hasattr(cls, "__visit_name__"): _generate_dispatch(cls) super(VisitableType, cls).__init__(clsname, bases, clsdict) def _generate_dispatch(cls): if "__visit_name__" in cls.__dict__: visit_name = cls.__visit_name__ if isinstance(visit_name, str): getter = operator.attrgetter("visit_%s" % visit_name) def _compiler_dispatch(self, visitor, **kw): try: meth = getter(visitor) except AttributeError: raise exc.UnsupportedCompilationError(visitor, cls) else: return meth(self, **kw) cls._compiler_dispatch = _compiler_dispatch class Visitable(util.with_metaclass(VisitableType, object)): passCopy the codeVisitable convention subclasses must provide the class attribute visit_NAME to bind the compilation method. Classes participating in SQL are inherited from Visitable:
class SchemaItem(SchemaEventTarget, visitors.Visitable):
__visit_name__ = "schema_item"
class MetaData(SchemaItem):
__visit_name__ = "metadata"
class Table(DialectKWArgs, SchemaItem, TableClause):
__visit_name__ = "table"
class Column(DialectKWArgs, SchemaItem, ColumnClause):
__visit_name__ = "column"
class TypeEngine(Visitable):
...
class Integer(_LookupExpressionAdapter, TypeEngine):
__visit_name__ = "integer"
Copy the codeMetaData is a collection of schemas that record all Table definitions. The _add_table function is used to add tables:
class MetaData(SchemaItem): def __init__( self, bind=None, reflect=False, schema=None, quote_schema=None, naming_convention=None, info=None, ): # table self.tables = util.immutabledict() self.schema = quoted_name(schema, quote_schema) self._schemas = set() def _add_table(self, name, schema, table): key = _get_table_key(name, schema) dict.__setitem__(self.tables, key, table) if schema: self._schemas.add(schema)Copy the codeTable is a collection of columns. Add it to the metadata:
class Table(DialectKWArgs, SchemaItem, TableClause): def __new__(cls, *args, **kw): name, metadata, args = args[0], args[1], Args [2:] schema = metadata.schema table = object.__new__(CLS) # table) table._init(name, metadata, *args, **kw) return table def _init(self, name, metadata, *args, **kwargs): super(Table, self).__init__( quoted_name(name, kwargs.pop("quote", None)) self.metadata = metadata self.schema = metadata. Schema # column self._columns = column Collection() self._init_items(*args) def _init_items(self, *args): # column for item in args: if item is not None: item._set_parent_with_dispatch(self)Copy the codeColumn is added to the colummns of the table using the following method:
class Column(DialectKWArgs, SchemaItem, ColumnClause):
def __init__(self, *args, **kwargs):
pass
def _set_parent(self, table):
table._columns.replace(self)
class ColumnCollection(util.OrderedProperties):
def replace(self, column):
...
self._data[column.key] = column
...
Copy the codeMetadata holds a table collection, and table holds a column collection. Let’s take a look at how this data structure can be converted into an SQL statement. The API is implemented through the metadata. create_all function:
class MetaData(SchemaItem):
def create_all(self, bind=None, tables=None, checkfirst=True):
bind._run_visitor(
ddl.SchemaGenerator, self, checkfirst=checkfirst, tables=tables
)
class Engine(Connectable, log.Identified):
def _run_visitor(
self, visitorcallable, element, connection=None, **kwargs
):
with self._optional_conn_ctx_manager(connection) as conn:
conn._run_visitor(visitorcallable, element, **kwargs)
class Connection(Connectable):
def _run_visitor(self, visitorcallable, element, **kwargs):
visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
Copy the codeDDL SchemaGenerator (traverse_single);
Class ClauseVisitor(object): def traverse_single(self, obj, **kw): def traverse_single(self, obj, **kw): meth = getattr(v, "visit_%s" % obj.__visit_name__, None) if meth: return meth(obj, **kw) @property def visitor_iterator(self): v = self while v: yield v v = getattr(v, "_next", None) class SchemaVisitor(ClauseVisitor): ... class DDLBase(SchemaVisitor): ... class SchemaGenerator(DDLBase): ...Copy the codeCreate meta, table, and columun
class SchemaGenerator(DDLBase): def visit_metadata(self, metadata): tables = list(metadata.tables.values()) collection = sort_tables_and_constraints( [t for t in tables if self._can_create_table(t)] ) for table, fkcs in collection: if table is not None: Self. traverse_single(table, create_OK =True, include_foreign_KEY_constraints = FKCS, _is_metadatA_operation =True, ) def visit_table( self, table, create_ok=False, include_foreign_key_constraints=None, _is_metadata_operation=False, ): for column in table.columns: if column.default is not None: # create column-DDLElement self.traverse_single(column.default) self.connection. Execute (# FMT: Create table-DDLElement CreateTable(table, DDLElement) include_foreign_key_constraints= # noqa include_foreign_key_constraints, ) # fmt: on )Copy the codeCreateTableDDLElement and CreateColumnDDLElement inheritance:
class _DDLCompiles(ClauseElement):
def _compiler(self, dialect, **kw):
return dialect.ddl_compiler(dialect, self, **kw)
class DDLElement(Executable, _DDLCompiles):
...
class _CreateDropBase(DDLElement):
...
class CreateTable(_CreateDropBase):
__visit_name__ = "create_table"
def __init__(
self, element, on=None, bind=None, include_foreign_key_constraints=None
):
super(CreateTable, self).__init__(element, on=on, bind=bind)
self.columns = [CreateColumn(column) for column in element.columns]
class CreateColumn(_DDLCompiles):
__visit_name__ = "create_column"
def __init__(self, element):
self.element = element
Copy the codeIn the end, these DDLelements are compiled into SQL statements in the DDLCompiler. CREATE TABLE is compiled like this:
def visit_create_table(self, create):
table = create.element
preparer = self.preparer
text = "\nCREATE "
if table._prefixes:
text += " ".join(table._prefixes) + " "
text += "TABLE " + preparer.format_table(table) + " "
create_table_suffix = self.create_table_suffix(table)
if create_table_suffix:
text += create_table_suffix + " "
text += "("
separator = "\n"
# if only one primary key, specify it along with the column
first_pk = False
for create_column in create.columns:
column = create_column.element
try:
processed = self.process(
create_column, first_pk=column.primary_key and not first_pk
)
if processed is not None:
text += separator
separator = ", \n"
text += "\t" + processed
if column.primary_key:
first_pk = True
except exc.CompileError as ce:
...
const = self.create_table_constraints(
table,
_include_foreign_key_constraints=create.include_foreign_key_constraints, # noqa
)
if const:
text += separator + "\t" + const
text += "\n)%s\n\n" % self.post_create_table(table)
return text
def visit_create_column(self, create, first_pk=False):
column = create.element
text = self.get_column_specification(column, first_pk=first_pk)
const = " ".join(
self.process(constraint) for constraint in column.constraints
)
if const:
text += " " + const
return text
Copy the codeIn the previous column introduction, we skipped over data types. We all know that SQL data types are different from Python data types. Here are some common SQL data types:
class TypeEngine(Visitable):
...
class Integer(_LookupExpressionAdapter, TypeEngine):
__visit_name__ = "integer"
...
class String(Concatenable, TypeEngine):
__visit_name__ = "string"
...
class CHAR(String):
__visit_name__ = "CHAR"
...
class VARCHAR(String):
__visit_name__ = "VARCHAR"
...
Copy the codeThe data type is compiled by GenericTypeCompiler:
class TypeCompiler(util.with_metaclass(util.EnsureKWArgType, object)):
def process(self, type_, **kw):
return type_._compiler_dispatch(self, **kw)
class GenericTypeCompiler(TypeCompiler):
def visit_INTEGER(self, type_, **kw):
return "INTEGER"
def visit_string(self, type_, **kw):
return self.visit_VARCHAR(type_, **kw)
def visit_VARCHAR(self, type_, **kw):
return self._render_string_type(type_, "VARCHAR")
def _render_string_type(self, type_, name):
text = name
if type_.length:
text += "(%d)" % type_.length
if type_.collation:
text += ' COLLATE "%s"' % type_.collation
return text
Copy the codeData Manipulation Language (DML) uses INSERTS to insert Data
The API for data insertion is provided by TableClause.
class TableClause(Immutable, FromClause):
@util.dependencies("sqlalchemy.sql.dml")
def insert(self, dml, values=None, inline=False, **kwargs):
return dml.Insert(self, values=values, inline=inline, **kwargs)
Copy the codeAn implementation of the Insert class is provided in DML:
class UpdateBase(
HasCTE, DialectKWArgs, HasPrefixes, Executable, ClauseElement
):
...
class ValuesBase(UpdateBase):
...
class Insert(ValuesBase):
__visit_name__ = "insert"
...
Copy the codeFollowing our DDL experience, we looked for the compilation method for insert statements in SQLCompiler:
class SQLCompiler(Compiled): def visit_insert(self, insert_stmt, asfrom=False, **kw): crud_params = crud._setup_crud_params( self, insert_stmt, crud.ISINSERT, **kw ) if insert_stmt._has_multi_parameters: crud_params_single = crud_params[0] else: crud_params_single = crud_params preparer = self.preparer supports_default_values = self.dialect.supports_default_values text = "INSERT " text += "INTO " table_text = preparer.format_table(insert_stmt.table) if crud_params_single or not supports_default_values: text += " (%s)" % ", ".join( [preparer.format_column(c[0]) for c in crud_params_single] ) ... if insert_stmt.select is not None: select_text = self.process(self._insert_from_select, **kw) if self.ctes and toplevel and self.dialect.cte_follows_insert: text += " %s%s" % (self._render_cte_clause(), select_text) else: text += " %s" % select_text elif not crud_params and supports_default_values: text += " DEFAULT VALUES" elif insert_stmt._has_multi_parameters: text += " VALUES %s" % ( ", ".join( "(%s)" % (", ".join(c[1] for c in crud_param_set)) for crud_param_set in crud_params ) ) else: text += " VALUES (%s)" % ", ".join([c[1] for c in crud_params]) return textCopy the codeYou can see that the INSERT statement is a concatenation of the INSERT object through a string template.
Data Query Language (DQL) uses SELECT to Query Data
Data query SELECT statements also have specific data structure SELECT, inheritance relationship as follows:
class SelectBase(HasCTE, Executable, FromClause):
...
class GenerativeSelect(SelectBase):
...
class Select(HasPrefixes, HasSuffixes, GenerativeSelect):
__visit_name__ = "select"
def __init__(
self,
columns=None,
whereclause=None,
from_obj=None,
distinct=False,
having=None,
correlate=True,
prefixes=None,
suffixes=None,
**kwargs
):
GenerativeSelect.__init__(self, **kwargs)
...
Copy the codeThe compilation statement for select is also in SQLCompiler:
class SQLCompiler(Compiled):
def visit_select(
self,
select,
asfrom=False,
parens=True,
fromhints=None,
compound_index=0,
nested_join_translation=False,
select_wraps_for=None,
lateral=False,
**kwargs
):
...
froms = self._setup_select_stack(select, entry, asfrom, lateral)
column_clause_args = kwargs.copy()
column_clause_args.update(
{"within_label_clause": False, "within_columns_clause": False}
)
text = "SELECT " # we're off to a good start !
text += self.get_select_precolumns(select, **kwargs)
# the actual list of columns to print in the SELECT column list.
inner_columns = [
c
for c in [
self._label_select_column(
select,
column,
populate_result_map,
asfrom,
column_clause_args,
name=name,
)
for name, column in select._columns_plus_names
]
if c is not None
]
...
text = self._compose_select_body(
text, select, inner_columns, froms, byfrom, kwargs
)
if select._statement_hints:
per_dialect = [
ht
for (dialect_name, ht) in select._statement_hints
if dialect_name in ("*", self.dialect.name)
]
if per_dialect:
text += " " + self.get_statement_hint_text(per_dialect)
if self.ctes and toplevel:
text = self._render_cte_clause() + text
if select._suffixes:
text += " " + self._generate_prefixes(
select, select._suffixes, **kwargs
)
self.stack.pop(-1)
if (asfrom or lateral) and parens:
return "(" + text + ")"
else:
return text
Copy the codeThe SELECT statement is also a string concatenation.
The ORM example
Orm is used in a slightly different way than Schema. Here is an example of orM:
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///:memory:', echo=True)
Model = declarative_base()
class User(Model):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
fullname = Column(String)
nickname = Column(String)
def __repr__(self):
return "<User(name='%s', fullname='%s', nickname='%s')>" % (
self.name, self.fullname, self.nickname)
Model.metadata.create_all(engine)
print("="*10)
Session = sessionmaker(bind=engine)
session = Session()
ed_user = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
session.add(ed_user)
session.commit()
print(ed_user.id)
result = engine.execute("select * from users")
for row in result:
print(row)
Copy the codeThe following table can be obtained by comparing schema and ORM:
Schema way | way orm engine, used for database connection | – create metadata, Used to manage schema | create Model to create the users Table Table | create User Model to the metadata to submit to the engine (create Table) | — – | | create session using the users insert data using insert data session
To sum up, there are two main differences:
- Orm does not display the schema for creating tables
- Orm data processing is done using sessions, not connections
Core Functions of Model
The Model class is created dynamically using Declarative_base:
class DeclarativeMeta(type):
def __init__(cls, classname, bases, dict_):
if "_decl_class_registry" not in cls.__dict__:
_as_declarative(cls, classname, cls.__dict__)
type.__init__(cls, classname, bases, dict_)
def __setattr__(cls, key, value):
_add_attribute(cls, key, value)
def __delattr__(cls, key):
_del_attribute(cls, key)
def declarative_base(
bind=None,
metadata=None,
mapper=None,
cls=object,
name="Base",
constructor=_declarative_constructor,
class_registry=None,
metaclass=DeclarativeMeta,
):
# 创建metadata
lcl_metadata = metadata or MetaData()
if class_registry is None:
class_registry = weakref.WeakValueDictionary()
bases = not isinstance(cls, tuple) and (cls,) or cls
class_dict = dict(
_decl_class_registry=class_registry, metadata=lcl_metadata
)
# 构造函数
if constructor:
class_dict["__init__"] = constructor
if mapper:
class_dict["__mapper_cls__"] = mapper
# class-meta
return metaclass(name, bases, class_dict)
Copy the codeHow to create classes dynamically is described in tips. Declarative_base mainly defines several features of the Model class:
- The constructor of the Model class
__init__Using _declarative_constructor - Subclasses of the Model class call _AS_declarative when they are constructed
- The model object is assigned using _add_attribute
We’ll start with the _declarative_constructor:
def _declarative_constructor(self, **kwargs):
cls_ = type(self)
for k in kwargs:
if not hasattr(cls_, k):
raise TypeError(
"%r is an invalid keyword argument for %s" % (k, cls_.__name__)
)
setattr(self, k, kwargs[k])
_declarative_constructor.__name__ = "__init__"
Copy the codeIt looks simple enough, but there is a validation conversion between the class and the object instance. Let’s start with a demo code:
class DummyModel(object): Name = ["dummy_model"] # Reference type A = DummyModel() B = DummyModel() Assert ID == id(B.name) == ID (dummyModel.name) a.name.append("a") assert id(a.name) == id(b.name) == id(DummyModel.name)Copy the codeDummyModel’s class attribute name is the same reference as the name attribute of the A object. If using the Model class:
Model = declarative_base() class UserModel(Model): __tablename__ = 'user' # Mandatory id = Column(Integer, Primary_key =True) # Mandatory field name = Column(String) C = UserModel() c.name = "c" d = UserModel() d.name = "D assert isinstance(UserModel.name, InstrumentedAttribute) assert isinstance(c.name, str) assert d.name == "d" assert id(c.name) ! = id(d.name) ! = id(UserModel.name)Copy the codeInstrumentedAttribute = “InstrumentedAttribute” > < span style = “line-height: 20px; line-height: 20px; This is one of the features of the ORM Model. The Model defines the format template, and objects are instantiated and converted into ordinary data.
Another feature of the Model is the implicit creation of a Table object, implemented in the _AS_declarative function via _MapperConfig
class _MapperConfig(object):
def setup_mapping(cls, cls_, classname, dict_):
cfg_cls = _MapperConfig
cfg_cls(cls_, classname, dict_)
def __init__(self, cls_, classname, dict_):
...
self._setup_table()
...
def _setup_table(self):
...
table_cls = Table
args, table_kw = (), {}
if table_args:
if isinstance(table_args, dict):
table_kw = table_args
elif isinstance(table_args, tuple):
if isinstance(table_args[-1], dict):
args, table_kw = table_args[0:-1], table_args[-1]
else:
args = table_args
autoload = dict_.get("__autoload__")
if autoload:
table_kw["autoload"] = True
cls.__table__ = table = table_cls(
tablename,
cls.metadata,
*(tuple(declared_columns) + tuple(args)),
**table_kw
)
...
Copy the codeColumn is implemented using the following function:
def _add_attribute(cls, key, value):
if "__mapper__" in cls.__dict__:
if isinstance(value, Column):
_undefer_column_name(key, value)
cls.__table__.append_column(value)
cls.__mapper__.add_property(key, value)
...
else:
type.__setattr__(cls, key, value)
Copy the codeIn this way, Model implicitly creates a Schema(Table). In actual use, you only need to use the Model class and do not care about the definition of the Schema.
The session source code is left for later analysis due to space and time constraints
summary
Sqlalchemy can be used at a low level in a way that provides SQL statements; Provide the definition schema mode for use at the sub-level; The orM implementation is provided at a high level, allowing applications to choose between different levels of API depending on the characteristics of the project.
When using a Schema, Metadata, tables, and columns are used to define the schema data structure, and the compiler is used to automatically convert the schema into legitimate SQL statements.
With ORM, a specific data model is created. Schema objects are implicitly created, and data is accessed in session mode.
A final review of sqlAlchemy’s architecture diagram:
tip
Sqlalchemy provides a way to create classes dynamically, mainly in Declarative_base and DeclarativeMeta. I made a class factory based on this implementation:
class DeclarativeMeta(type): def __init__(cls, klass_name, bases, dict_): print("class_init", klass_name, bases, dict_) type.__init__(cls, klass_name, bases, dict_) def get_attr(self, key): print("getattr", self, key) return self.__dict__[key] def constructor(self, *args, **kwargs): print("constructor", self, args, kwargs) for k, v in kwargs.items(): setattr(self, k, v) def dynamic_class(name): class_dict = { "__init__": constructor, "__getattr__": get_attr } return DeclarativeMeta(name, (object,), class_dict) DummyModel = dynamic_class("Dummy") dummy = DummyModel(1, name="hello", age=18) print(dummy, type(dummy), dummy.name, dummy.age) # class_init Dummy (<class 'object'>,) {'__init__': <function test_dynamic_class.<locals>.constructor at 0x7f898827ef70>, '__getattr__': <function test_dynamic_class.<locals>.get_attr at 0x7f89882105e0>} # constructor <sample.Dummy object at 0x7f89882a5820> (1,) {'name': 'hello', 'age': 18} # <sample.Dummy object at 0x7f89882a5820> <class 'sample.Dummy'> hello 18Copy the codeIn this example, I dynamically created a DummyModel class, type(dummy), which is named dummy. The available constructor for this class accepts both name and age attributes. This creation is somewhat similar to collections.namedtuple.
A little feeling
Sqlalchemy source code is very complex, a total of one month after preparation, the formation of two documents only involves the core process and usage, details are missing more, there is a chance to continue to read. During this month, I overcame my impatience that I was busy with work and had no time to write. Overcame reading into the dilemma, once wanted to give up the psychological barrier; Overcame the shame of using the saved manuscript to replace the manuscript because the deadline was approaching and the manuscript was still a prototype; Overcome the note-taking software glitches, lost written documents, completely rewritten frustration. Overcome these difficulties, and finally to complete, there is great psychological satisfaction. Of course, the biggest harvest is to have a preliminary understanding of ORM middleware, and I hope that the ORM process can be helpful to you. If we get your support, we will be more satisfied with ♥️.