def __init__(self, request):
self.request = request
self.entity_id = request.matchdict.get('id')
self.entity = None
if self.entity_id:
from stalker import SimpleEntity
from sqlalchemy.exc import DataError
try:
self.entity = SimpleEntity.query\
.filter(SimpleEntity.id == self.entity_id).first()
except DataError:
self.entity = None
if not self.entity:
from pyramid.exceptions import HTTPNotFound
raise HTTPNotFound('Entity not found!')
python类DataError()的实例源码
def insert(employee: dict) -> dict:
"""
Insert a new employee profile
:return: created employee profile
"""
logger.info('New employee profile: {profile}'.format(profile=employee))
if Employee.query.filter(Employee.name == employee.get('name')).first():
logger.warning(warning.ALREADY_EXISTS)
abort(400, {'message': warning.ALREADY_EXISTS})
else:
try:
employee.pop('id', None)
employee['registered'] = datetime.utcnow()
db.session.add(Employee(**employee))
db.session.commit()
logger.info('Profile saved!')
except DataError:
abort(400, {'message': warning.INVALID_DATA_TYPE})
return NoContent, 200
def update(employee_id: int, employee: dict) -> dict:
"""
Update employee profile by id
:param employee_id: employee id
:param employee: employee profile
:return: updated employee profile
"""
try:
logger.info('Employee id that want to be updated: {id}'.format(id=employee_id))
selected_employee = Employee.query.filter(Employee.id == employee_id).first_or_404()
selected_employee.update(**employee)
db.session.commit()
logger.info('Updated!')
return NoContent, 201
except DataError:
abort(400, {'message': warning.INVALID_DATA_TYPE})
def create_or_update_time_spent(task_id, person_id, date, duration, add=False):
try:
time_spent = TimeSpent.get_by(
task_id=task_id,
person_id=person_id,
date=date
)
except DataError:
raise WrongDateFormatException
if time_spent is not None:
if add:
time_spent.update({"duration": time_spent.duration + duration})
else:
time_spent.update({"duration": duration})
else:
time_spent = TimeSpent.create(
task_id=task_id,
person_id=person_id,
date=date,
duration=duration
)
return time_spent.serialize()
def login_user(cls, name, password):
session = SessionManager.Session()
try:
user = session.query(User).filter(User.name == name).one()
if check_password_hash(user.password, password):
credential = cls(user)
SessionManager.Session.remove()
return credential
else:
raise ClientError(ClientError.LOGIN_FAIL)
except NoResultFound:
raise ClientError(ClientError.LOGIN_FAIL)
except DataError:
raise ClientError(ClientError.LOGIN_FAIL)
except ClientError as error:
raise error
except Exception as error:
raise ServerError(error.message)
finally:
SessionManager.Session.remove()
def setEmployeePosition():
'''This method is used to store the new position for the employee
Example : POST /api/v1/employeepostions
{"name" : "Cook", "description" : "THis is desription"}
'''
with SessionManager(Session) as session:
try:
name = request.json['name']
description = request.json.get('description', 'NA')
sql_pos = EmployeePosition(name=name, description=description)
session.add(sql_pos)
session.commit()
return jsonify(post_envelop(200, data = request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Value : {0} already exists'.format(name)))
except:
return jsonify(error_envelop(400, 'UnknownError', 'Error need to be identified'))
def test_blogpost_title_length(self):
"""Test BLOGPOST model title length has a limit"""
self.configure_fixtures()
valid_title = 'a' * 255
invalid_title = 'a' * 256
blogpost = Blogpost(title=valid_title, body="body", project=self.project)
db.session.add(blogpost)
assert_not_raises(DataError, db.session.commit)
blogpost.title = invalid_title
assert_raises(DataError, db.session.commit)
def _check_if_pod_exists(pod_id):
user = KubeUtils.get_current_user()
try:
pod = Pod.filter(Pod.owner_id == user.id, Pod.id == pod_id).first()
if pod is None:
raise PodNotFound
except DataError:
# pod_id is not uuid
raise PodNotFound
def test_invalid_date(self):
self.insert_employee()
payload = {'birthdate': 'test'}
res = self.test_app.put(self.url, headers=self.headers, data=json.dumps(payload))
self.assertEqual(res.status_code, 400)
from sqlalchemy.exc import DataError
self.assertRaises(DataError)
def test_invalid_date(self):
payload = {'name': 'budi2', 'gender': Employee.Gender.MALE,
'birthdate': 'test', 'active': True}
res = self.test_app.post('/employee', headers=self.headers, data=json.dumps(payload))
self.assertEqual(res.status_code, 400)
from sqlalchemy.exc import DataError
self.assertRaises(DataError)
def test_country_name_overflow_error(session, country_data):
"""Country 003: Verify error if country name exceeds field length."""
too_long_country = Countries(**country_data['overflow'])
with pytest.raises(DataError):
session.add(too_long_country)
session.commit()
def test_competition_name_overflow_error(session, comp_data):
"""Competition 003: Verify error if competition name exceeds field length."""
overflow_record = Competitions(**comp_data['overflow'])
with pytest.raises(DataError):
session.add(overflow_record)
session.commit()
def registerReferenceSet(self, guid, assembly_id, source_accessions=None, description=None, references=OrderedDict()):
"""
ReferenceSet registration for MAF occurs from an assembly config file. See hg19.json for example.
ReferenceSet registration for VCF occurs from reading VCF contig tags in header.
Requires assembly ids and guids to be unique.
"""
referenceSet = self.session.query(ReferenceSet).filter(
or_(ReferenceSet.assembly_id == assembly_id, ReferenceSet.guid == guid))\
.first()
if referenceSet is None:
try:
referenceSet = ReferenceSet(
guid=guid, assembly_id=assembly_id, description=description)
self.session.add(referenceSet)
self.session.commit()
except exc.DataError as e:
self.session.rollback()
raise ValueError("{0} : {1} ".format(str(e), guid))
if len(references) > 0:
# use pyvcf like ordered dict to avoiding having to specify
# reference order manually
refs = sortReferences(references)
for ref in refs:
self.registerReference(
str(uuid.uuid4()), referenceSet.id, ref, refs[ref].length)
return referenceSet
def registerReference(self, guid, reference_set_id, name, length):
"""
Registers a Reference. Most often called by registerReferenceSet.
Requires a Reference name be unique for all references in a reference set
"""
# contig MT is same as contig M, in meta db we will always use M to be
# consistent
if name == 'MT':
name = 'M'
reference = self.session.query(Reference).filter(
and_(Reference.reference_set_id == reference_set_id,Reference.name == name))\
.first()
if reference is None:
try:
reference = Reference(
name=name,
reference_set_id=reference_set_id,
length=length,
guid=guid
)
self.session.add(reference)
self.session.commit()
except exc.DataError as e:
self.session.rollback()
raise ValueError("{0} : {1} ".format(str(e), guid))
return reference
def registerWorkspace(self, guid, name):
"""
Registers a workspace.
Workspace name is the path to the workspace directory. This is assumed unique per metadb instance.
"""
# if the name ends with a / then remove it from the name.
# This is done only for consistency in workspace name
# since users could have / or not for the workspace.
name = name.rstrip('/')
workspace = self.session.query(Workspace).filter(
and_(Workspace.name == name))\
.first()
if workspace is None:
try:
workspace = Workspace(
guid=guid,
name=name
)
self.session.add(workspace)
self.session.commit()
except exc.DataError as e:
self.session.rollback()
raise ValueError("{0} : {1} ".format(str(e), guid))
return workspace
def registerDBArray(self, guid, reference_set_id, workspace_id, name):
"""
Registers a DBArray.
An array is unique named folder in a unique workspace path and a given reference id.
"""
# array is a unique set of workspace, array, and reference set
# association
dbarray = self.session.query(DBArray) .filter(
and_(DBArray.reference_set_id == reference_set_id,\
DBArray.workspace_id == workspace_id,\
DBArray.name == name))\
.first()
if dbarray is None:
try:
dbarray = DBArray(
guid=guid,
reference_set_id=reference_set_id,
workspace_id=workspace_id,
name=name
)
self.session.add(dbarray)
self.session.commit()
except exc.DataError as e:
self.session.rollback()
raise ValueError("{0} : {1} ".format(str(e), guid))
return dbarray
def registerVariantSet(self, guid, reference_set_id, dataset_id=None, metadata=None):
"""
Register variant set.
"""
referenceSet = self.session.query(ReferenceSet).filter(
ReferenceSet.id == reference_set_id)\
.first()
if referenceSet is None:
raise ValueError(
"ReferenceSet must be registered before registering this VariantSet : {0} ".format(reference_set_id))
variantSet = self.session.query(
VariantSet).filter(VariantSet.guid == guid)\
.first()
if variantSet is None:
try:
variantSet = VariantSet(
guid=guid,
reference_set_id=reference_set_id,
dataset_id=dataset_id,
variant_set_metadata=metadata
)
self.session.add(variantSet)
self.session.commit()
except exc.DataError as e:
self.session.rollback()
raise ValueError("{0} : {1} ".format(str(e), guid))
return variantSet
def registerIndividual(self, guid, name, info=None):
"""
Registration of an individual requires a guid and a name.
Name can be None to support retrival from registerSample
"""
individual = self.session.query(Individual).filter(
or_(Individual.guid == guid, Individual.name == name))\
.first()
if individual is None:
try:
individual = Individual(
name=name,
guid=guid,
info=info,
record_update_time=strftime("%Y-%m-%d %H:%M:%S%S.%S%S%S"),
record_create_time=strftime("%Y-%m-%d %H:%M:%S%S.%S%S%S")
)
self.session.add(individual)
self.session.commit()
except (exc.DataError, exc.IntegrityError) as e:
self.session.rollback()
raise ValueError("{0} : {1} : {2} ".format(str(e), guid, name))
return individual
def test_MetaTooBig(session, db_library, request):
lid = db_library.id
v = "!" * (Meta.VERSION_LIMIT + 1)
session.add(Meta(version=v))
if "sqlite" in request.keywords:
session.commit()
assert session.query(Meta).filter_by(version=v).one()
else:
with pytest.raises(DataError):
session.commit()
def test_LibraryTooBig(session, db_library, request):
lid = db_library.id
v = "!" * (Library.NAME_LIMIT + 1)
session.add(Library(name=v))
if "sqlite" in request.keywords:
session.commit()
assert session.query(Library).filter_by(name=v).one()
else:
with pytest.raises(DataError):
session.commit()
def test_ImageType(session, db_library, request, mp3audiofile):
img = Image(type="Beyond", size=666,
md5="48a6de198166ad9806f1a2172f2947df",
mime_type="img/jpg", description="Iceburn",
data=b"\xde\xad\xbe\xef")
session.add(img)
if "sqlite" in request.keywords:
with pytest.raises(IntegrityError):
session.commit()
else:
assert "postgresql" in request.keywords
with pytest.raises(DataError):
session.commit()
def setVat():
'''This function is used to store the new vat in the database
Example : POST /api/v1/vats HTTP/1.1
{"name" : "food", "value":34}
Result : {
"meta": {
"code": 200,
"message": "Created Successfully"
}
}
'''
with SessionManager(Session) as session:
try:
name = request.json['name']
value = request.json['value']
vat = Vat(name=name, value=value)
session.add(vat)
session.commit()
return jsonify(post_envelop(200, data = request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Value : {0} already exists'.format(name)))
except:
return jsonify(error_envelop(400,'UnknownError','Error need to be identified'))
def setServiceCharge():
'''This function is used to store the new servicecharge in the database
Example : POST /api/v1/service charges HTTP/1.1
{"name" : "Service Charge", "value":10}
Result : {
"meta": {
"code": 200,
"message": "Created Successfully"
}
}
'''
with SessionManager(Session) as session:
try:
name = request.json['name']
value = request.json['value']
service_charge = ServiceCharge(name=name, value=value)
session.add(service_charge)
session.commit()
return jsonify(post_envelop(200, data = request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Value : {0} already exists'.format(name)))
except:
return jsonify(error_envelop(400,'UnknownError','Error need to be identified'))
def setPayment():
'''This function is used to store the new payment in the database
Example : POST /api/v1/payments HTTP/1.1
{ "p_type": " onsite"}
Result : {
"meta": {
"code": 200,
"message": "Created Successfully"
}
}
'''
with SessionManager(Session) as session:
try:
p_type = request.json['p_type']
payment = Payment(p_type=p_type)
session.add(payment)
session.commit()
return jsonify(post_envelop(200, data = request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Value : {0} already exists'.format(p_type)))
except:
return jsonify(error_envelop(400,'UnknownError','Error need to be identified'))
def setDineTable():
'''This function is used to store the new DineTable in the database
Example : POST /api/v1/dinetables HTTP/1.1
{ "capicity": 4, "alias" : "Table1", "status" : "empty"}
Result : {
"meta": {
"code": 200,
"message": "Created Successfully"
}
}
'''
with SessionManager(Session) as session:
try:
capacity = request.json['capacity']
alias = request.json['alias']
status = request.json.get('status', 'empty')
dine_table = DineTable(capacity=capacity, alias=alias, status=status)
session.add(dine_table)
session.commit()
return jsonify(post_envelop(200, data = request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Value : {0} already exists'.format(alias)))
except:
return jsonify(error_envelop(400,'UnknownError','Error need to be identified'))
def setMembership():
'''This function is used to store the new Membership in the database
Example : POST /api/v1/memberships HTTP/1.1
{ "m_type": "general", "discount" : 10, "description" : "some description"}
Result : {
"meta": {
"code": 200,
"message": "Created Successfully"
}
}
'''
with SessionManager(Session) as session:
try:
m_type = request.json['m_type']
discount = request.json['discount']
if not 0 <= int(discount) < 100:
return jsonify(error_envelop(400, 'DataError', 'Enter the valid discount amount (0 to 100)'))
description = request.json.get('description', 'No Description Available')
membership = Membership(m_type=m_type, discount=discount, description=description)
session.add(membership)
session.commit()
return jsonify(post_envelop(200, data = request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Value : {0} already exists'.format(m_type)))
except:
return jsonify(error_envelop(400,'UnknownError','Error need to be identified'))
def updateMembership(m_id):
''' PUT /api/v1/dientables/6 HTTP/1.1
{"alias" : "new table name", "capacity" : 3}
Result : {
"data": {
"capacity": 3,
"alias" : "new table name"
},
"meta": {
"code": 200,
"message": "Updated Successfully"
}
}
'''
with SessionManager(Session) as session:
try:
sql_membership = session.query(Membership).filter(Membership.id == m_id).one()
sql_membership.m_type = request.json.get('m_type', sql_membership.m_type)
sql_membership.discount = request.json.get('discount', sql_membership.discount)
#check weather the discount is between 0 and 100
if not 0 <= int(sql_membership.discount) < 100:
return jsonify(error_envelop(400, 'DataError', 'Enter the valid discount amount (0 to 100)'))
sql_membership.description = request.json.get('description', sql_membership.description)
session.commit()
return jsonify(update_envelop(200, data=request.json))
except IntegrityError:
# if name already exsits in database
return jsonify(error_envelop(400, 'Integrity Error','Name already Exists'))
except NoResultFound:
return jsonify(error_envelop(404, 'ValueError', 'Id : {0} not found'.format(m_id)))
return jsonify(error_envelop(100, 'UnknownError', 'UnknownError Found'))
#now the item is succesfulluy updated
def setCustomersByMembership(m_id):
'''This function is used to set the customer based on membership_id as a foriegn key'''
with SessionManager(Session) as session:
try:
first_name = request.json['first_name']
last_name = request.json['last_name']
middle_name = request.json.get('middle_name', 'NA')
contact_number = request.json.get('contact_number', 'NA')
address = request.json.get('address', 'NA')
gender = request.json['gender']
age = request.json['age']
email = request.json.get('email', 'NA')
if not 6 <= int(age) < 99:
return jsonify(error_envelop(400, "Age Error", "Please enter the age between 6 and 99 "))
c = Customer(first_name = first_name,
last_name = last_name,
contact_number = contact_number,
address = address,
gender = gender,
age =age,
membership_id=m_id,
middle_name=middle_name,
email=email)
session.add(c)
session.commit()
return jsonify(post_envelop(200, data=request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Violates foreign key ({0}) constraint'.format(m_id)))
except:
return jsonify(error_envelop(400,'UnknownError','Error need to be identified'))
def setBill():
with SessionManager(Session) as session:
try:
total_price = request.json['total_price']
on_site = request.json.get('on_site')
customer_id = request.json['customer_id']
dinetable_id = request.json['dinetable_id']
employee_id = request.json['employee_id']
payment_id = request.json['payment_id']
vat_id = request.json['vat_id']
service_charge_id = request.json['service_charge_id']
bill_description = request.json.get('bill_description', 'NA')
item_orders = request.json['item_orders']
if len(item_orders) < 1:
return jsonify(error_envelop(400, 'BillError', 'Please enter atleast one item!!'))
#if there is more than one elements in bills then create a new bill object
bill = Bill(total_price=total_price,
bill_description=bill_description,
on_site=on_site,
customer_id=customer_id,
dinetable_id=dinetable_id,
employee_id=employee_id,
payment_id=payment_id,
vat_id=vat_id,
service_charge_id=service_charge_id)
session.add(bill)
for item_order in item_orders:
item_id = item_order['item_id']
quantity = item_order['quantity']
order_price = item_order['order_price']
session.add(ItemOrder(item_id=item_id,
quantity=quantity,
order_price=order_price,
bill=bill))
session.commit()
return jsonify(post_envelop(200, data = request.json))
except DataError: #this excepyion might probably occur if the value key has a value of non integer
return jsonify(error_envelop(400, 'DataError', 'Use the correct value'))
except IntegrityError:
return jsonify(error_envelop(400, 'IntegrityError','Foreign Key violations. Use the correct id'))
except:
return jsonify(error_envelop(400, 'UnknownError', 'Error need to be identified'))
