sqlalchemy 动态分表

约定不等于承诺〃 2022-01-15 05:11 443阅读 0赞

[为什么80%的码农都做不了架构师？>>> ][80_]  ![hot3.png][]

最近项目需要长期添加一些记录, 随着数据的增加, 单表已经出现瓶颈, 决定探索下sqlalchemy.

./lib/database/bash.py

# coding: utf-8
    
    from sqlalchemy import create_engine, Table
    from sqlalchemy.orm import mapper
    from sqlalchemy.orm import sessionmaker, scoped_session
    
    
    class SqlalchemyFactory(object):
        # _session = None
        _instance = None
    
        def __new__(cls, *args, **kw):
            if not cls._instance:
                cls._instance = super(SqlalchemyFactory, cls).__new__(cls)
            return cls._instance
    
        def __init__(self, uri, echo, encoding, metadata):
            kwargs = dict(echo=echo, encoding=encoding, convert_unicode=True)
            self._metadata = metadata
            self._engine = create_engine(uri, **kwargs)
            self._session_maker = sessionmaker(
                bind=self._engine,
                autoflush=False,
                autocommit=False
            )
    
        def session(self):
            # if not self._session:
            session = scoped_session(self._session_maker)
            _session = session()
            _session.text_factory = str
            return _session
    
        def create_table(self, table_name, *columns):
            table = Table(table_name, self._metadata, *columns, extend_existing=True)
            if not table.exists(bind=self._engine):
                table.create(bind=self._engine, checkfirst=True)
            return table
    
        def drop_table(self, table_name):
            table = Table(table_name, self._metadata, extend_existing=True)
            if not table.exists(bind=self._engine):
                return
            table.drop(bind=self._engine)
            return table
    
        def model(self, table_name, class_name, *columns):
            try:
                # tables = Base.metadata.tables
                # if table_name in tables:
                #     return tables[table_name]
                table = self.create_table(table_name, *columns)
                # Base.metadata.reflect(self._engine)
                model = type(class_name, (object,), dict())
                mapper(model, table)
                # Base.metadata.clear()
                return model
            except Exception as e:
                self.drop_table(table_name)
                print('SqlalchemyFactory model error: %s' % e)

./lib/database/test.py

# coding: utf-8
    
    import config
    from datetime import datetime
    from lib.logger import logger
    from sqlalchemy import Column, Integer, String, DateTime, Text
    from sqlalchemy.ext.declarative import declarative_base
    from lib.database.base import SqlalchemyFactory
    
    Base = declarative_base()
    
    
    class TestFactory(SqlalchemyFactory):
        _instance = None
    
        def __new__(cls, *args, **kw):
            if not cls._instance:
                cls._instance = super(TestFactory, cls).__new__(cls)
            return cls._instance
    
        def __init__(self, uri=config.TEST_URI, echo=config.STORAGE_ECHO, encoding=config.STORAGE_ENCODING):
            super(TestFactory, self).__init__(uri, echo, encoding, Base.metadata)
    
        def __init_db__(self):
            Base.metadata.create_all(self._engine)
    
    
    
    class Test(Base):
        __tablename__ = 'test'
        __table_args__ = {'mysql_charset': 'utf8'}
        id = Column('id', Integer, primary_key=True, autoincrement=True)
        query = Column('query', String(512), nullable=False, default=None)
        data = Column('data', Text, nullable=True)
        ip = Column('ip', String(20), nullable=True)
        create_time = Column('create_time', DateTime, nullable=False, default=datetime.now)
    
        @staticmethod
        def table_name(suffix=None):
            return '%s_%s' % (test.__tablename__, suffix if suffix else '')
    
        @staticmethod
        def class_name(suffix=None):
            return '%s%s' % (test.__name__, suffix if suffix else '')
    
        @staticmethod
        def columns():
            return [
                Column('id', Integer, primary_key=True, autoincrement=True),
                Column('query', String(512), nullable=False, default=None),
                Column('data', Text, nullable=True),
                Column('ip', String(20), nullable=True),
                Column('create_time', DateTime, nullable=False, default=datetime.now)
            ]
    
        @staticmethod
        def table(factory=None):
            if not factory:
                factory = TestFactory()
            date_time = datetime.now().strftime('%y%m')
            columns = Test.columns()
            class_name = Test.class_name(date_time)
            table_name = Test.table_name(date_time)
            return factory.model(table_name, class_name, *columns)
    
        @staticmethod
        def insert_(query, data, ip):
            factory = TestFactory()
            session = factory.session()
            try:
                test_class = Test.table(factory)
                test = test_class()
                test.query = query
                test.data = data
                test.ip = ip
                session.add(test)
                session.commit()
            except Exception as e:
                logger.error('database Test insert error: %s' % e)
            finally:
                session.close()
    
    
    if __name__ == '__main__':
        import time
        start = time.time()
        for i in range(100):
            tmp_start = time.time()
            Test.insert_('test', 'test', '127.0.0.1')
            print(i, time.time() - tmp_start)
        end = time.time()
        print(end - start)

ok. 已经完成对test的分表. 可以执行下test.py文件. 按照月份拆分数据表.

如想马上看到结果, 可以吧把时间放到分钟或者秒级

转载于:https://my.oschina.net/yehun/blog/3047627

[80_]: https://my.oschina.net/u/2663968/blog/3051541
[hot3.png]: /images/20220115/5762730d8dfc41deaf13239ba6229eae.png