個人筆記,利用python call api及DB操作
API
import requests
from requests.auth import HTTPBasicAuth
class APIClient:
def __init__(self, base_url, userid, pwd):
self.base_url = base_url
self.auth = HTTPBasicAuth(userid, pwd)
def post(self, endpoint,parames,headers=None,verify=False):
url = f"{self.base_url}/{endpoint}"
response = requests.post(url, auth=self.auth, json=parames
,headers=headers
, verify=verify)
if response.status_code == 200:
return response.json()
else:
response.raise_for_status()
def get(self,endpoint,parames,headers=None,verify=False):
url = f"{self.base_url}/{endpoint}"
response = requests.get(url, auth=self.auth, headers=headers
,params=parames, verify=verify)
if response.status_code == 200:
return response.json()
else:
response.raise_for_status()
DB操作
import pyodbc
import pandas as pd
import re
class DBUtil:
def __init__(self, connectionString):
self.connectionString=connectionString
def getConnectin(self):
conn = pyodbc.connect(self.connectionString)
return conn
def getConnectin(self,connectionString):
'''
連線字串
'''
conn = pyodbc.connect(connectionString)
return conn
def closeConnection(self,conn):
'''
關閉Connection
'''
try:
conn.close()
except Exception as e:
print("close db error:",e)
def execute_query(self,conn, sql_query):
"""
Execute a SQL query on the database.
"""
result = pd.read_sql_query(sql_query,conn)
return result
def prepare_commandstring(self,query_template, params):
"""
:param query_template: 有命名参数的查詢模板
:param params: 参数字典
:return: (最終的查詢字串, 參數touple)
SELECT * FROM your_table_name
WHERE column1 = {param1} AND column2 = {param2} OR column3 = {param1}
"""
# 使用字符串格式化生成最终的查询
query = query_template.format(**{key: '?' for key in params.keys()})
# 使用正则表达式找到所有的占位符
param_order = re.findall(r'\{(\w+)\}', query_template)
# print(param_order)
# 生成参数元组,确保每个占位符都有对应的参数值
param_values = tuple(params[key] for key in param_order)
# 返回最终的查询字符串和参数元组
return query, param_values
打雜打久了,就變成打雜妹
程式寫久了,就變成老乞丐