[PYTHON] Alchimie SQL modifiée

Une version modifiée de SQL Alchemy, le mappeur O / R standard pour Pyramid et Flask.

Problème de syntaxe SQL Alchemy

Démon rénové

J'ai essayé de le remodeler pour qu'il puisse être décrit simplement. En termes de vitesse, la connexion à la base de données est connectée en pool à l'aide de ThreadLocalStorage, elle fonctionne donc à environ 8 fois la vitesse d'inutilisation. Dans l'environnement local, la vitesse d'une requête de sélection a été augmentée de 8 ms à 1 ms.

Exemple d'utilisation après la modification magique de SQLAlchemy

# -*- coding: utf-8 -*-
from module.book import Book

# select
book1 = Book.get(1)
books = Book.objects().filter(Book.price==2160).all()

# insert
book_rye = Book(pk=None,
                title="The catcher in the rye",
                price=1000,
                publish="J. D. Salinger",
                published="")
book_rye = Book.insert(book_rye)
print(book_rye.id)

# update
book_rye.price = 1200
book_rye.save()

# delete
book_rye.delete()

book.py

# -*- coding: utf-8 -*-
from sqlalchemy import Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from module.db.base import DBBaseMixin

Base = declarative_base()


class Book(DBBaseMixin, Base):
    title = Column('title', String(200))
    price = Column('price', Integer)
    publish = Column('publish', String(200))
    published = Column('published', String(200))

    def __init__(self, pk, title, price, publish, published):
        self.pk = pk
        self.title = title
        self.price = price
        self.publish = publish
        self.published = published

base_mixin.py

# -*- coding: utf-8 -*-
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base, declared_attr
import re
from utils.db import get_db_session

Base = declarative_base()


def camel_to_snake(s):
    """
    >>> convert('CamelCase')
    'camel_case'
    >>> convert('CamelCamelCase')
    'camel_camel_case'
    >>> convert('Camel2Camel2Case')
    'camel2_camel2_case'
    >>> convert('getHTTPResponseCode')
    'get_http_response_code'
    >>> convert('get2HTTPResponseCode')
    'get2_http_response_code'
    >>> convert('HTTPResponseCode')
    'http_response_code'
    >>> convert('HTTPResponseCodeXYZ')
    'http_response_code_xyz'
    :param s: str
    :return: str
    """
    s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', s)
    return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()


class DBBaseMixin(object):
    id = Column('id', Integer, primary_key=True)

    def __repr__(self):
        return '<{0}.{1} object at {2}>'.format(
            self.__module__, type(self).__name__, hex(id(self)))

    @declared_attr
    def __tablename__(cls):
        return camel_to_snake(cls.__name__)

    @classmethod
    def objects(cls):
        """
        :rtype : sqlalchemy.orm.query.Query
        """
        return get_db_session().query(cls)

    @classmethod
    def session(cls):
        return get_db_session()

    @classmethod
    def get(cls, pk):
        """
        :param pk: int
        :rtype: cls
        """
        return cls.objects().get(pk)

    @classmethod
    def insert(cls, obj):
        """
        :param obj: cls
        :rtype: cls
        """
        cls.session().add(obj)
        cls.session().commit()
        return obj

    @classmethod
    def bulk_insert(cls, objs):
        """
        :param objs: list[cls]
        :rtype: list[cls]
        """
        cls.session().add_all(objs)
        cls.session().commit()
        return objs

    def delete(self):
        session = self.__class__.session()
        session.query(self.__class__).filter(self.__class__.id==self.id).delete()
        session.commit()

    def save(self):
        """
        :return:self
        """
        session = self.__class__.session()
        session.add(self)
        session.commit()
        return self

db.py

# -*- coding: utf-8 -*-
import threading
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

tls = threading.local()


def get_db_session():
    """
Générer et réutiliser des sessions SQL Alchemy DB
    :rtype : scoped_session
    """
    if hasattr(tls, "db_session"):
        return tls.db_session

    #Génération de session DB
    engine = get_db_engine()
    db_session = scoped_session(sessionmaker(autocommit=False,
                                             autoflush=False,
                                             bind=engine))
    tls.db_session = db_session
    return db_session


def get_db_engine():
    db_user = 'root'
    db_host = '127.0.0.1'
    db_name = 'Flask'
    db_path = 'mysql://{}@{}/{}'.format(db_user, db_host, db_name)
    engine = create_engine(db_path, encoding='utf-8', pool_size=5)
    return engine

Un problème assez fatal avec ce code

Il ne prend pas en charge la mise à jour par le traitement des transactions DB. Je pense que cela peut être géré si vous pouvez générer une session avec la clause with et enregistrer (transaction_session = session). Bien sûr, la validation est exécutée avec __exit__ dans la clause with.

Pseudo code pour les transactions DB

#Prix d'échange
with commit_on_success as session:
  # select for update
  book1 = Book.get(1, transaction_session=session, for_update=True)
  book2 = Book.get(2, transaction_session=session, for_update=True)

  # exchange price
  _tmp_price = book1.price
  book1.price = book2.price
  book2.price = _tmp_price
  
  # update
  book1.save(transaction_session=session)
  book2.save(transaction_session=session)