本文整理汇总了Python中pymongo.MongoClient.replace_one方法的典型用法代码示例。如果您正苦于以下问题：Python MongoClient.replace_one方法的具体用法？Python MongoClient.replace_one怎么用？Python MongoClient.replace_one使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在类pymongo.MongoClient的用法示例。
在下文中一共展示了MongoClient.replace_one方法的4个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: __init__
# 需要导入模块: from pymongo import MongoClient [as 别名]
# 或者: from pymongo.MongoClient import replace_one [as 别名]
class SolutionManager:
""" Manage solutions.
The form of a solution
name: human readable name of a solution
alg: algorithm used; Ex.) logistic regression
dataset: training/testing data as 2-d array shaped as [# of samples, # of features]
situations: a list of situation labels for each sample in the dataset
serialized_obj: serialized created model instance
desc: description of a solution
accuracy: accuracy determined after verifying a model with 70% training dataset and 30% testing dataset
domain: domain of a solution
owner: id of a user who create the solution
visibility: public or private
auth_users: users who are accessible to the solution even though the visibility is private
"""
def __init__(self):
self.db = MongoClient().infaas.solutions
def add_solution(self, solution):
# if the entered solution is not valid, return False.
if not self._verify_solution(solution):
return False
res = self.db.insert_one(solution)
if res.acknowledged:
return True
return False
def get_solutions(self, name=None):
query = {}
if name:
query.update({"name": name})
solutions = self.db.find(query)
return solutions
def update_solution(self, solution):
# if the entered solution is not valid, return False.
if not self._verify_solution(solution):
return False
query = {"_id": solution.get("_id")}
new = {k: v for k, v in solution.items() if k in ATTRIBUTES}
res = self.db.replace_one(query, new)
return True if res.modified_count > 0 else False
def delete_solution(self, solution):
query = {"_id": solution.get("_id")}
res = self.db.delete_one(query)
return True if res.deleted_count > 0 else False
@staticmethod
def _verify_solution(solution):
if not solution:
return False
return False示例2: __init__
# 需要导入模块: from pymongo import MongoClient [as 别名]
# 或者: from pymongo.MongoClient import replace_one [as 别名]
class DomainManager:
""" Manage domains.
The form of a domain
name: human readable name of a domain; it should be unique.
desc: description of a domain
features: a list of features; [{'name', 'desc'}, ...]
situations: a list of possible situations; possible inference results; [{'name', 'desc'}, ...]
"""
def __init__(self):
self.db = MongoClient().INFaaS.domains
def add_domain(self, domain):
verified = self._verify_domain(domain)
if not verified or not set(domain.keys()) >= ATTRIBUTES:
raise Exception(constants.MSG_INVALID_PARAMS)
if 'owner' in domain:
domain['owner'] = ObjectId(domain.get('owner'))
res = self.db.insert_one(domain)
if res.acknowledged:
return True
return False
def get_domains(self, name=None):
query = {}
if name:
query.update({'name': name})
domains = self.db.find(query)
return domains
def update_domain(self, domain):
# if the entered solution is not valid, return False.
if not self._verify_domain(domain):
return False
query = {'_id': domain.get('_id')}
new = {k: v for k, v in domain.items() if k in ATTRIBUTES}
res = self.db.replace_one(query, new)
return True if res.modified_count > 0 else False
def delete_domain(self, domain):
query = {'_id': domain.get('_id')}
res = self.db.delete_one(query)
return True if res.deleted_count > 0 else False
@staticmethod
def _verify_domain(domain):
if not domain:
return False
return True示例3: __init__
# 需要导入模块: from pymongo import MongoClient [as 别名]
# 或者: from pymongo.MongoClient import replace_one [as 别名]
class UserManager:
""" Manage users.
"""
def __init__(self):
self.db = MongoClient().INFaaS.users
def add_user(self, user):
if user.keys() >= REQ_ATTRIBUTES:
res = self.db.insert_one({k: v for k, v in user.items() if k in ATTRIBUTES})
return res.acknowledged
return False
def get_user(self, email=None, password=None, apikey=None, status=None):
query = {}
if email:
query.update({'email': email})
if password:
query.update({'password': password})
if apikey:
query.update({'apikey': apikey})
if status:
query.update({'status': status})
return self.db.find_one(query)
def update_user(self, user):
# if the entered solution is not valid, return False.
if not self._verify_user(user):
return False
query = {'_id': user.get('_id')}
new = {k: v for k, v in user.items() if k in ATTRIBUTES}
res = self.db.replace_one(query, new)
return True if res.modified_count > 0 else False
def generate_apikey(self, user):
email = user.get('email')
if not email:
return False
apikey = str(uuid.uuid5(uuid.NAMESPACE_URL, email))
self.db.update_one({'email': email}, {'$set': {'apikey': apikey}})
@staticmethod
def _verify_user(user):
if not user:
return False
return True示例4: store_portmap
# 需要导入模块: from pymongo import MongoClient [as 别名]
# 或者: from pymongo.MongoClient import replace_one [as 别名]
def store_portmap(pm):
coll=MongoClient(dburl).lsfportmap.lsfportmap
return coll.replace_one(pmquery, {'name': 'portmap',
'portmap': set2lst(pm)})本文标签属性：
示例：示例图
代码：代码大全可复制
Python：python什么意思
MongoClient：net
replace_one：replace_one
