作者:HuPangpang_ | 来源:互联网 | 2023-05-18 18:19
#encodingutf-8@version:??@author:xq@contact:xiaoq_xiaoq@163.com@file:gpsDataMinStd
#encoding = utf-8
"""
@version:??
@author: xq
@contact:xiaoq_xiaoq@163.com
@file: gpsDataMinStd.py
@time: 2017/10/12 13:13
"""
from pymongo import MongoClient
# 连接mongodb获取数据
class dbApi(object):
def __init__(self, ipAddress, dbPort):
self.dbAddress = ipAddress self.dbPort = dbPort def dbConn(self,dbDatabase,dbCollection,user,password):
'''
# 建立MongoDB数据库连接
:param dbDatabase: 数据库名
:return: 连接对象
'''
url = self.dbAddress+":"+self.dbPort
mongodburi = "mongodb://"+user+":"+password+"@"+self.dbAddress+":"+self.dbPort+"/"+dbDatabase+"?authMechanism=SCRAM-SHA-1"
self.client = MongoClient(mongodburi)
# 连接所需数据库,loct为数据库名
db = self.client.loct
if (db != None): # 如果数据库连接成功则连接所用集合
collection = db.Collect
if (collection != None): # 如果选择集合成功则返回集合
return collection
else: # 如果选择集合失败则给出提示并退出
print("集合 dataCollect选择失败...")
exit(0)
else: # 如果选择数据库失败则给出提示并退出
print("Connect to database fail......")
exit(0)
def getData(self, collection):
'''
# 从数据库获取数据
:param collection: 数据库集合对象
:return: 数据列表
'''
#collection = dbConn()
if (collection == None):
print("没有连接到mongodb......")
exit(0)
else:
dbData = collection.find({"deviceId": "5000105792"})
gpsData = []
for item in dbData:
gpsData.append(item)
return gpsData
def dbClose(self):
'''
# 关闭数据库
:return:
'''
if (self.client != None):
self.client.close()
else:
print("没有连接到mongodb")
exit(0)
#测试
def main():
mongodbConn = dbApi('19.2.28.1', '100')
dbCollection = mongodbConn.dbConn('loct','Collect','cfi','cfi')
gpsData = mongodbConn.getData(dbCollection)
mongodbConn.dbClose()
print(gpsData)
if __name__ == '__main__':
main()