consolidator
/
OraWorker.py
49 строк · 1.9 Кб
1import oracledb as ora
2import pandas as pd
3
4class OraWorker(object):
5
6def __init__(self,user:str,password:str,dsn:str,client:str) -> None:
7try:
8ora.init_oracle_client(lib_dir=client)
9self.connection = ora.connect(dsn=dsn,user=user,password=password)
10self.cursor = self.connection.cursor()
11except: raise
12
13def ora_version(self)->str:
14try:
15res = []
16for row in self.cursor.execute("SELECT * FROM v$version"):
17res.append(f"{row}")
18# for row in self.cursor.execute("SELECT * FROM v$instance"):
19# res.append(f"{row}")
20# for row in self.cursor.execute("select sysdate as dt, 1 as ver from dual"):
21# res.append(f"{row[0]}")
22return "\n".join(res)
23except :raise
24
25def product_exists(self,product:str)->bool:
26try:
27# print(product)
28self.cursor.execute("select count(1) from ione.items where state='A' and code_name=:code",[product])
29result=self.cursor.fetchone()[0]
30return result==1
31except: raise
32
33def product_check(self,product:str,nomenclature:str)->tuple:
34try:
35sql="""select
36(select nvl(max(m.scp_product_name),'нет') from scm_fact.product_map_sap_scp m where m.sap_product_code=:NOM) nom_to_up,
37(select count(1) from ione.items where state='A' and code_name=:UP) exists_in_up
38from dual"""
39self.cursor.execute(sql,[nomenclature,product])
40record = self.cursor.fetchone()
41correct = 1 if record[0]==product else 0
42return (record[0],record[1],correct)
43except Exception as exp : return(f"{exp}",0,0)
44
45def close(self):
46try:
47self.cursor.close()
48self.connection.close()
49except: raise