From c3df44d5794c33b1a742bb0c8606382d6ba34363 Mon Sep 17 00:00:00 2001 From: old-tom <892955278@qq.com> Date: Wed, 5 Apr 2023 15:12:01 +0800 Subject: [PATCH] feat:first commit --- fudb/connectors/__init__.py | 7 + fudb/connectors/connector_factory.py | 32 +++ fudb/connectors/dialect/__init__.py | 7 + fudb/connectors/dialect/abs_connector.py | 25 +++ fudb/connectors/dialect/dialect_connector.py | 35 ++++ fudb/dbapis/__init__.py | 7 + fudb/dbapis/fu_collection.py | 24 +++ fudb/dbapis/fu_db_api.py | 195 +++++++++++++++++++ requirements.txt | Bin 0 -> 196 bytes test/__init__.py | 7 + test/test_connector_factory.py | 17 ++ test/test_dialect_connector.py | 44 +++++ test/test_oracle_connector.py | 20 ++ 13 files changed, 420 insertions(+) create mode 100644 fudb/connectors/__init__.py create mode 100644 fudb/connectors/connector_factory.py create mode 100644 fudb/connectors/dialect/__init__.py create mode 100644 fudb/connectors/dialect/abs_connector.py create mode 100644 fudb/connectors/dialect/dialect_connector.py create mode 100644 fudb/dbapis/__init__.py create mode 100644 fudb/dbapis/fu_collection.py create mode 100644 fudb/dbapis/fu_db_api.py create mode 100644 requirements.txt create mode 100644 test/__init__.py create mode 100644 test/test_connector_factory.py create mode 100644 test/test_dialect_connector.py create mode 100644 test/test_oracle_connector.py diff --git a/fudb/connectors/__init__.py b/fudb/connectors/__init__.py new file mode 100644 index 0000000..a228065 --- /dev/null +++ b/fudb/connectors/__init__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/4 16:02 +# @Author : old tom +# @File : __init__.py.py +# @Project : futool-db +# @Desc : 数据库连接器 diff --git a/fudb/connectors/connector_factory.py b/fudb/connectors/connector_factory.py new file mode 100644 index 0000000..776be22 --- /dev/null +++ b/fudb/connectors/connector_factory.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/4 18:33 +# @Author : old tom +# @File : connector_factory.py +# @Project : futool-db +# @Desc : 连接器工厂 + +from fudb.connectors.dialect.dialect_connector import OracleConnector, PostgresqlConnector +from urllib.parse import quote_plus as urlquote + + +class ConnFactory(object): + """ + 数据库连接器工厂 + """ + CONNECTION_CONTAINER = { + 'oracle': OracleConnector, + 'postgresql': PostgresqlConnector + } + + def __init__(self, db_type, user, password, host, port, database): + self.db_type = db_type + # urlquote 用于处理密码中的特殊字符例如@ + self.connector = self.CONNECTION_CONTAINER[self.db_type](user, urlquote(password), host, port, database) + + def get_conn(self): + """ + 获取连接 + :return: + """ + return self.connector.get_conn() diff --git a/fudb/connectors/dialect/__init__.py b/fudb/connectors/dialect/__init__.py new file mode 100644 index 0000000..c5abc5d --- /dev/null +++ b/fudb/connectors/dialect/__init__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/4 18:54 +# @Author : old tom +# @File : __init__.py.py +# @Project : futool-db +# @Desc : diff --git a/fudb/connectors/dialect/abs_connector.py b/fudb/connectors/dialect/abs_connector.py new file mode 100644 index 0000000..d5a9744 --- /dev/null +++ b/fudb/connectors/dialect/abs_connector.py @@ -0,0 +1,25 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/4 16:04 +# @Author : old tom +# @File : abs_connector.py +# @Project : futool-db +# @Desc : 抽象层 + +import abc +from sqlalchemy import create_engine + + +class CommonConnector(metaclass=abc.ABCMeta): + + def __init__(self, db_conf: str): + # 初始化 + self.engine = create_engine(db_conf, pool_size=15, pool_recycle=3600) + + @abc.abstractmethod + def get_conn(self): + """ + 获取连接 + :return: + """ + pass diff --git a/fudb/connectors/dialect/dialect_connector.py b/fudb/connectors/dialect/dialect_connector.py new file mode 100644 index 0000000..e954f92 --- /dev/null +++ b/fudb/connectors/dialect/dialect_connector.py @@ -0,0 +1,35 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/4 18:55 +# @Author : old tom +# @File : dialect_connector.py +# @Project : futool-db +# @Desc : +from fudb.connectors.dialect.abs_connector import CommonConnector + + +class OracleConnector(CommonConnector): + """ + oracle 连接器,目前仅支持sever_name方式,SID方式待开发 + """ + + DSN = 'oracle+cx_oracle://{0}:{1}@{2}:{3}/?service_name={4}' + + def __init__(self, user, password, host, port=1521, server_name='orcl'): + super().__init__(self.DSN.format(user, password, host, port, server_name)) + + def get_conn(self): + return self.engine.connect() + + +class PostgresqlConnector(CommonConnector): + """ + pg连接器 + """ + PG_DIALECT = 'postgresql+psycopg2://{0}:{1}@{2}:{3}/{4}' + + def __init__(self, user, password, host, port=5432, database='postgres'): + super().__init__(self.PG_DIALECT.format(user, password, host, port, database)) + + def get_conn(self): + return self.engine.connect() diff --git a/fudb/dbapis/__init__.py b/fudb/dbapis/__init__.py new file mode 100644 index 0000000..30f1242 --- /dev/null +++ b/fudb/dbapis/__init__.py @@ -0,0 +1,7 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/4 16:03 +# @Author : old tom +# @File : __init__.py.py +# @Project : futool-db +# @Desc : 操作数据库API diff --git a/fudb/dbapis/fu_collection.py b/fudb/dbapis/fu_collection.py new file mode 100644 index 0000000..af3ad4b --- /dev/null +++ b/fudb/dbapis/fu_collection.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/5 11:26 +# @Author : old tom +# @File : fu_collection.py +# @Project : futool-db +# @Desc : 集合类工具 + +def split_coll(data: [], part_size=5): + """ + 分割集合 + :param data: + :param part_size: + :return: + """ + rt = [] + if len(data) <= part_size: + rt.append(data) + else: + rt.append(data[0:part_size]) + for j, d in enumerate(data): + if j > 0 and j % part_size == 0: + rt.append(data[j:j + part_size]) + return rt diff --git a/fudb/dbapis/fu_db_api.py b/fudb/dbapis/fu_db_api.py new file mode 100644 index 0000000..66d275a --- /dev/null +++ b/fudb/dbapis/fu_db_api.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# @Time : 2023/4/4 17:29 +# @Author : old tom +# @File : fu_db_api.py +# @Project : futool-db +# @Desc : + +from sqlalchemy import Connection, text, CursorResult +from fudb.dbapis.fu_collection import split_coll +from psycopg2.extras import execute_values + + +class SqlExecuteError(Exception): + def __init__(self, msg=''): + Exception.__init__(self, msg) + + +def _select(conn: Connection, sql) -> CursorResult: + """ + 自动关闭连接 + :param conn: + :return: + """ + try: + return conn.execute(text(sql)) + finally: + conn.close() + + +def _execute_with_tx(conn: Connection, sql): + """ + 带事务执行SQL + :return: + """ + try: + conn.begin() + rt = conn.execute(text(sql)) + conn.commit() + return rt.rowcount + except Exception as e: + conn.rollback() + raise SqlExecuteError(msg=f'sql [{sql}] 执行失败,开始回滚,e={e}') + finally: + conn.close() + + +def select_one(conn: Connection, sql): + """ + 查询一个 + :param conn: + :param sql: + :return: + """ + return _select(conn, sql).fetchone() + + +def select_all(conn: Connection, sql): + """ + 查询全部 + :param conn: + :param sql: + :return: + """ + return _select(conn, sql).fetchall() + + +def count(conn: Connection, table): + """ + 统计数据量 + :param conn: + :param table: + :return: + """ + count_tpl = f'select count(1) from {table}' + return select_one(conn, count_tpl)[0] + + +def execute_update(conn: Connection, sql): + """ + 带事务执行,可用于insert update delete 语句 + :param conn: + :param sql: + :return: 受影响的行数，与java-jdbc的execute_update返回true|false相似,可用于判断是否执行成功 + """ + return _execute_with_tx(conn, sql) + + +def batch_insert(conn: Connection, db_type, sql_tpl, data, batch_size=1500): + """ + 批量插入 + :param conn: 数据库连接 + :param batch_size: 每次插入量 + :param db_type: 数据库类型 + :param sql_tpl: insert into t1 (f1,f2,f3) values %s + :param data: [(1,'tom',29),(2,'jack',30)] + :return: + """ + handler = BatchInsertHandler(db_type, sql_tpl, data, batch_size) + insert_sqls = handler.build_insert() + # 整个插入都在一个事务内 + row_count = 0 + try: + conn.begin() + for sql_set in insert_sqls: + rt = conn.execute(text(sql_set)) + row_count += rt.rowcount + conn.commit() + return row_count + except Exception as e: + conn.rollback() + raise SqlExecuteError(msg=f"批量插入异常,e={e}") + finally: + conn.close() + + +class BatchInsertHandler(object): + """ + 批量插入处理器 + oracle : + insert all + into oracle_table ( id, code ) values( 1 , '1' ) + into oracle_table ( id, code ) values( 2 , '2' ) + into oracle_table ( id, code ) values( 3 , '3' ) + into oracle_table ( id, code ) values( 4 , '4' ) + select 1 from dual ; + postgresql and mysql + into oracle_table ( id, code ) values( 1 , '1' ),( 2 , '2' ),( 3 , '3' ) + + """ + + BUILD_INSERT = { + 'oracle': 'build_oracle_insert', + 'postgresql': 'build_pg_insert', + 'mysql': 'build_mysql_insert' + } + + class NotSupportError(Exception): + def __init__(self, msg=''): + Exception.__init__(self, msg) + + def __init__(self, db_type, sql_tpl, data, batch_size): + """ + :param db_type: 数据库类型 + :param sql_tpl: insert into t1 (f1,f2,f3) values %s + :param data: [(1,'tom',29),(2,'jack',30)] + :param batch_size: + """ + if db_type not in ['oracle', 'postgresql']: + raise self.NotSupportError() + self.db_type = db_type + self.sql_tpl = sql_tpl + self.data = data + self.batch_size = batch_size + + def gen_batch_sql(self) -> []: + """ + 生成批量插入SQL + :return: + """ + pass + + def _split_data(self): + return split_coll(self.data, self.batch_size) + + def build_insert(self): + data_set = self._split_data() + sql_set = [] + for part in data_set: + sql_set.append(getattr(self, self.BUILD_INSERT[self.db_type])(part)) + return sql_set + + def build_oracle_insert(self, data_set): + begin = 'insert all \r ' + for ds in data_set: + val = '(' + for ele in ds: + val += "'" + ele + "'," if isinstance(ele, str) else str(ele) + ',' + val = val[0:-1] + ')' + begin += (self.sql_tpl.replace('%s', val) + ' \r ') + end = 'select 1 from dual' + return begin + end + + def build_pg_insert(self, data_set): + vals = '' + for ds in data_set: + val = '(' + for ele in ds: + val += "'" + ele + "'," if isinstance(ele, str) else str(ele) + ',' + val = val[0:-1] + ')' + vals += val + ',' + return self.sql_tpl.replace('%s', vals[0:-1]) + + def build_mysql_insert(self, data_set): + return self.build_pg_insert(data_set) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000000000000000000000000000000000000..ebb51d66de299fe3d43a8b7fb0a5da772860ffd1 GIT binary patch literal 196 zcmX|*u?oUK5JX=s_$dZC6}7Oi7ew*_B1edVnj?2b{CRa8h%DR8yq($4J2EiiNY91H zoraxeL&ZWnXUc?DZlihHFD8QBieMs