extended_declarative_base¶

Extend the power of declarative base.

class sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase(**kwargs)[source]¶

Provide additional method.

Example:

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base, ExtendedBase):
    ... do what you do with sqlalchemy ORM

中文文档

提供了三个快捷函数, 分别用于获得列表形式的 primary key names, fields, values

ExtendedBase.pk_names()
ExtendedBase.pk_fields()
ExtendedBase.pk_values()

另外提供了三个快捷函数, 专门针对只有一个 primary key 的情况, 分别用于获得单个形式的 primary key name, field, value.

ExtendedBase.id_field_name()
ExtendedBase.id_field()
ExtendedBase.id_field_value()

所有参数包括 engine_or_session 的函数需返回

所有的 insert / update
所有的 select 相关的 method 返回的不是 ResultProxy, 因为有 engine_or_session
这个参数如果输入是 engine, 用于执行的 session 都是临时对象, 离开了这个 method, session 将被摧毁. 而返回的 Result 是跟当前的 session 绑定相关的, session 一旦被关闭, Result 理应不进行任何后续操作. 所以建议全部返回所有结果的列表而不是迭代器.

classmethod pk_names() → Tuple[str][source]¶: Primary key column name list.

classmethod pk_fields() → Tuple[sqlalchemy.orm.attributes.InstrumentedAttribute][source]¶

Primary key columns instance. For example:

class User(Base):
    id = Column(..., primary_key=True)
    name = Column(...)

User.pk_fields() # (User.id,)

Return type: tuple

pk_values() → tuple[source]¶

Primary key values

Return type: tuple

classmethod id_field_name() → str[source]¶: If only one primary_key, then return the name of it. Otherwise, raise ValueError.

classmethod id_field() → sqlalchemy.orm.attributes.InstrumentedAttribute[source]¶: If only one primary_key, then return the Class.field name object. Otherwise, raise ValueError.

id_field_value()[source]¶: If only one primary_key, then return the value of primary key. Otherwise, raise ValueError

classmethod keys() → List[str][source]¶

return list of all declared columns.

Return type: List[str]

values() → list[source]¶: return list of value of all declared columns.

items() → List[Tuple[str, Any]][source]¶: return list of pair of name and value of all declared columns.

to_dict(include_null=True) → Dict[str, Any][source]¶

Convert to dict.

Return type: dict

to_OrderedDict(include_null: bool = True) → collections.OrderedDict[source]¶: Convert to OrderedDict.

glance(_verbose: bool = True)[source]¶

Print itself, only display attributes defined in ExtendedBase._settings_major_attrs

Parameters: _verbose – internal param for unit testing

absorb(other: sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase, ignore_none: bool = True) → sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase[source]¶

For attributes of others that value is not None, assign it to self.

中文文档

将另一个文档中的数据更新到本条文档。当且仅当数据值不为None时。

revise(data: dict, ignore_none: bool = True) → sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase[source]¶

Revise attributes value with dictionary data.

中文文档

将一个字典中的数据更新到本条文档. 当且仅当数据值不为 None 时.

classmethod by_pk(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session], id_: Union[Any, List[Any], Tuple])[source]¶

Get one object by primary_key values.

Examples:

class User(Base):
    id = Column(Integer, primary_key)
    name = Column(String)

with Session(engine) as session:
    session.add(User(id=1, name="Alice")
    session.commit()

# User(id=1, name="Alice")
print(User.by_pk(1, engine))
print(User.by_pk((1,), engine))
print(User.by_pk([1,), engine))

with Session(engine) as session:
    print(User.by_pk(1, session))
    print(User.by_pk((1,), session))
    print(User.by_pk([1,), session))

中文文档

一个简单的语法糖, 允许用户直接用 primary key column 的值访问单个对象.

classmethod by_sql(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session], sql: Union[str, sqlalchemy.sql.elements.TextClause]) → List[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase][source]¶

Query with sql statement or texture sql.

Examples:

class User(Base):
    id = Column(Integer, primary_key)
    name = Column(String)

with Session(engine) as session:
    user_list = [
        User(id=1, name="Alice"),
        User(id=2, name="Bob"),
        User(id=3, name="Cathy"),
    ]
    session.add_all(user_list)
    session.commit()

results = User.by_sql(
    "SELECT * FROM extended_declarative_base_user",
    engine,
)

# [User(id=1, name="Alice"), User(id=2, name="Bob"), User(id=3, name="Cathy")]
print(results)

中文文档

一个简单的语法糖, 允许用户直接用 SQL 的字符串进行查询.

classmethod smart_insert(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session], obj_or_objs: Union[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase, List[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase]], minimal_size: int = 5, _op_counter: int = 0, _insert_counter: int = 0) → Tuple[int, int][source]¶

An optimized Insert strategy. :param minimal_size: internal bulk size for each attempts :param _op_counter: number of successful bulk INSERT sql invoked :param _insert_counter: number of successfully inserted objects.

Returns: number of bulk INSERT sql invoked. Usually it is greatly smaller than len(data). and also return the number of successfully inserted objects.

Warning

This operation is not atomic, if you force stop the program, then it could be only partially completed

中文文档

在Insert中, 如果已经预知不会出现IntegrityError, 那么使用Bulk Insert的速度要远远快于逐条Insert。而如果无法预知, 那么我们采用如下策略:

尝试Bulk Insert, Bulk Insert由于在结束前不Commit, 所以速度很快。
如果失败了, 那么对数据的条数开平方根, 进行分包, 然后对每个包重复该逻辑。
若还是尝试失败, 则继续分包, 当分包的大小小于一定数量时, 则使用逐条插入。

直到成功为止。

该 Insert 策略在内存上需要额外的 sqrt(n) 的开销, 跟原数据相比体积很小。但时间上是各种情况下平均最优的。

1.4 以后的重要变化: session 变得更聪明了.

classmethod update_all(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session], obj_or_objs: Union[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase, List[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase]], include_null: bool = True, upsert: bool = False) → Tuple[int, int][source]¶

The sqlalchemy.crud.updating.update_all() function in ORM syntax.

This operation IS NOT ATOMIC. It is a greedy operation, trying to update as much as it can.

Parameters

engine_or_session – an engine created by``sqlalchemy.create_engine``.
obj_or_objs – single object or list of object
include_null – update those None value field or not
upsert – if True, then do insert also.

Returns

number of row been changed

classmethod upsert_all(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session], obj_or_objs: Union[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase, List[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase]], include_null: bool = True) → Tuple[int, int][source]¶

The sqlalchemy.crud.updating.upsert_all() function in ORM syntax.

Parameters

engine_or_session – an engine created by``sqlalchemy.create_engine``.
obj_or_objs – single object or list of object
include_null – update those None value field or not

Returns

number of row been changed

classmethod delete_all(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session])[source]¶

Delete all data in this table.

TODO: add a boolean flag for cascade remove

classmethod count_all(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session]) → int[source]¶: Return number of rows in this table.

classmethod select_all(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session]) → List[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase][source]¶

classmethod random_sample(engine_or_session: Union[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session], limit: Optional[int] = None, perc: Optional[int] = None) → List[sqlalchemy_mate.orm.extended_declarative_base.ExtendedBase][source]¶

Return random ORM instance.

Return type: List[ExtendedBase]