python使用装饰器执行sql

mac2024-01-26  41

1 取数据(函数形式)

库:sqlalchemy(连接数据库)+ pandas(取数据)

1.1不用装饰器

如下,首先建立连接池,然后定义函数完成取数据的操作。可以完成任务,但缺点是重复代码太多。

import sqlalchemy import pandas as pd # 建立数据库连接池 key = sqlalchemy.create_engine("oracle://test:123456@127.0.0.1:1521/muscle") def get_emp(key): sq = ''' select * from scott.emp ''' # 建立连接 + 执行sql + 返回数据(DataFrame) with key.connect() as con: res = pd.read_sql_query(sq, con) return res def get_some_thing_else(key): sq = ''' select * from sometable ''' # 建立连接 + 执行sql + 返回数据(DataFrame) with key.connect() as con: res = pd.read_sql_query(sq, con) return res

1.2 使用装饰器

1.2.1 使用不带参数的装饰器

搭配方式:装饰器(内置连接,连接数据库执行sql)+ 函数(定义和返回sql)。 好处是显而易见的,被装饰的函数实际上只用于定义sql,而取数据的过程完全通过装饰器来实现。但问题是一旦完成装饰器的定义,数据连接就确定了,这样显然不够灵活。

import sqlalchemy import pandas as pd def exe_sql(func): ''' 执行sql取数据,返回pd.DataFrame :param func: :return: ''' key = sqlalchemy.create_engine("oracle://test:123456@127.0.0.1:1521/muscle") def wrapper(): sq = func() with key.connect() as con: res = pd.read_sql_query(sq, con) return res return wrapper @exe_sql def get_emp(): sq = ''' select * from scott.emp ''' return sq @exe_sql def get_some_thing_else(): sq = ''' select * from sometable ''' return sq res1 = get_emp() res2 = get_some_thing_else()

1.2.2 使用带参数的装饰器

搭配方式:装饰器(数据库连接作为参数,连接数据库执行sql)+ 函数(定义和返回sql)。 类似于在不带参数的装饰器外再套一层,此时数据连接作为参数传入,在使用时可以灵活的进行数据库的切换。

key = sqlalchemy.create_engine("oracle://test:123456@127.0.0.1:1521/muscle") def exe_sql2(key): ''' 执行sql取数据,返回pd.DataFrame :param func: :return: ''' def decorator(func): def wrapper(): sq = func() with key.connect() as con: res = pd.read_sql_query(sq, con) return res return wrapper return decorator @exe_sql2(key) def get_emp(): sq = ''' select * from scott.emp ''' return sq

2 取数据(类形式)

在复杂一些的场景下,也可以使用类的方式来定义和整合获数据抽取的过程。

库:sqlalchemy(连接数据库)+ pandas(取数据)

2.1不使用装饰器

问题和函数形式的一样,代码重复。

class GetScott(): def get_emp(self, key): sq = ''' select * from scott.emp ''' with key.connect() as con: res = pd.read_sql_query(sq, con) return res def get_some_thing_else(self, key): sq = ''' select * from sometable ''' with key.connect() as con: res = pd.read_sql_query(sq, con) return res scott = GetScott() scott.get_emp()

2.2使用装饰器

由于实例方法会默认传入自身作为第一个参数,因此修饰的时候需要用到’*args。如下装饰器可以用来修饰实例方法。但是这样的方式无法灵活的使用实例本身的属性

def exe_sql3(func): ''' 执行sql取数据,返回pd.DataFrame :param func: :return: ''' key = sqlalchemy.create_engine("oracle://test:123456@127.0.0.1:1521/muscle") def wrapper(*args): # args[0] 等同于 self 参数 sq = func(*args) with key.connect() as con: res = pd.read_sql_query(sq, con) return res return wrapper class GetScott(): @exe_sql3 def get_emp(cls): sq = ''' select * from scott.emp ''' return sq scott = GetScott() scott.get_emp()

2.3使用实例属性

由于实例方法通过self参数访问实例属性,那么修饰实例方法的装饰器只要获得self的引用就可以访问实例属性,因而可以将数据库连接作为实例属性进行赋值,而后通过装饰器进行调用。

def exe_sql4(key : str): ''' 修饰实例方法,通过传入属性名称调用实例属性 :param key: 实例属性名称 :return: ''' def decorator(func): # @functools.wraps(func) def wrapper(*args, **kwargs): self = args[0] # 实例本身 skey = getattr(self, key) sq = func(*args, **kwargs) with skey.connect() as con: res = pd.read_sql_query(sq, con) return res return wrapper return decorator class GetScott(): def __init__(self, key) -> None: self.key = key @exe_sql4('key') # 传入需要使用的属性的名称 def get_emp(cls): sq = ''' select * from scott.emp ''' return sq con = sqlalchemy.create_engine("oracle://test:123456@127.0.0.1:1521/muscle") scott = GetScott(con) scott.get_emp()

3 直接执行sql

通过pandas可以实现批量取数据,但是无法通过执行update等sql语句直接变更数据表,此类操作只能通过sqlalchemy实现。

3.1 修饰一般方法

def exe_sql5(key, issilent=False): ''' 创建连接 + 创建事务 + 执行sql + commit :param key: 数据库连接 :param issilent: :return: ''' def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): sq = func(*args, **kwargs) with key.connect() as con: # 建立连接 with con.begin() as cbg: # 创建事务 con.execute(sq) cbg.commit() if not issilent: print('sql exe successfully') return wrapper return decorator @exe_sql5(key) def update_emp(ename, sal): sq = ''' UPDATE scott.emp SET sal = %s WHERE ename = %s '''%(sal, ename) return sq update_emp('SMITH', 1800)

3.2 修饰类方法

def exe_sql6(key: str, issilent=False): ''' 创建连接 + 创建事务 + 执行sql + commit :param key: :param issilent: :return: ''' def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): self = args[0] sq = func(*args, **kwargs) skey = getattr(self, key) with skey.connect() as con: with con.begin() as cbg: con.execute(sq) cbg.commit() if not issilent: print('sql exe successfully') return wrapper return decorator
最新回复(0)