Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class Office(Model):
class Meta:
table_name = 'OfficeModel'
office_id = NumberAttribute(hash_key=True)
address = Location()
employees = ListAttribute(of=OfficeEmployeeMap)
class BooleanModel(Model):
class Meta:
table_name = 'BooleanTable'
user_name = UnicodeAttribute(hash_key=True)
is_human = BooleanAttribute()
class TreeLeaf2(MapAttribute):
value = NumberAttribute()
class TreeLeaf1(MapAttribute):
value = NumberAttribute()
left = TreeLeaf2()
right = TreeLeaf2()
class TreeLeaf(MapAttribute):
value = NumberAttribute()
left = TreeLeaf1()
right = TreeLeaf1()
class Meta:
region = 'us-east-1'
table_name = 'user'
user_id = NumberAttribute(hash_key=True)
class BankStatement(Model):
class Meta:
region = 'us-east-1'
table_name = 'statement'
user_id = NumberAttribute(hash_key=True)
balance = NumberAttribute(default=0)
active = BooleanAttribute(default=True)
class LineItem(Model):
class Meta:
region = 'us-east-1'
table_name = 'line-item'
user_id = NumberAttribute(hash_key=True)
created_at = UTCDateTimeAttribute(range_key=True, default=datetime.now())
amount = NumberAttribute()
currency = UnicodeAttribute()
class DifferentRegion(Model):
event_type = UnicodeAttribute(null=True)
event_project_id = UnicodeAttribute(null=True)
event_company_id = UnicodeAttribute(null=True)
event_company_name = UnicodeAttribute(null=True)
event_company_name_lower = UnicodeAttribute(null=True)
event_project_name = UnicodeAttribute(null=True)
event_project_name_lower = UnicodeAttribute(null=True)
event_user_name = UnicodeAttribute(null=True)
event_time = UTCDateTimeAttribute(default=datetime.datetime.now())
event_time_epoch = NumberAttribute(default=int(time.time()))
event_data = UnicodeAttribute(null=True)
event_date = UnicodeAttribute(null=True)
event_project_external_id = UnicodeAttribute(null=True)
event_date_and_contains_pii = UnicodeAttribute(null=True)
company_id_external_project_id = UnicodeAttribute(null=True)
contains_pii = BooleanAttribute(null=True)
user_id_index = EventUserIndex()
event_type_index = EventTypeIndex()
class Event(model_interfaces.Event):
"""
ORM-agnostic wrapper for the DynamoDB Event model.
"""
def __init__(
self,
event_id=None,
event_type=None,
user_id=None,
event_project_id=None,
event_company_id=None,
class Meta:
"""Meta class for Project."""
table_name = "cla-{}-projects".format(stage)
if stage == "local":
host = "http://localhost:8000"
project_id = UnicodeAttribute(hash_key=True)
project_external_id = UnicodeAttribute()
project_name = UnicodeAttribute()
project_name_lower = UnicodeAttribute(null=True)
project_individual_documents = ListAttribute(of=DocumentModel, default=[])
project_corporate_documents = ListAttribute(of=DocumentModel, default=[])
project_member_documents = ListAttribute(of=DocumentModel, default=[])
project_icla_enabled = BooleanAttribute(default=True)
project_ccla_enabled = BooleanAttribute(default=True)
project_ccla_requires_icla_signature = BooleanAttribute(default=False)
project_external_id_index = ExternalProjectIndex()
project_acl = UnicodeSetAttribute(default=set())
class Project(model_interfaces.Project): # pylint: disable=too-many-public-methods
"""
ORM-agnostic wrapper for the DynamoDB Project model.
"""
def __init__(
self,
project_id=None,
project_external_id=None,
project_name=None,
project_name_lower=None,
from datetime import datetime
from pynamodb.attributes import UnicodeAttribute, BooleanAttribute, UTCDateTimeAttribute
from pynamodb.models import Model
class TodoModel(Model):
class Meta:
read_capacity_units = 1
write_capacity_units = 1
region = 'localhost'
host = 'http://localhost:8000'
todo_id = UnicodeAttribute(hash_key=True, null=False)
text = UnicodeAttribute(null=False)
checked = BooleanAttribute(null=False, default=False)
created_at = UTCDateTimeAttribute(null=False, default=datetime.utcnow)
updated_at = UTCDateTimeAttribute(null=False)
def save(self, conditional_operator=None, **expected_values):
self.updated_at = datetime.utcnow()
super().save()
def __iter__(self):
for name, attr in self._get_attributes().items():
yield name, attr.serialize(getattr(self, name))
@staticmethod
def setup_model(model, region, table_name, is_remote=False):
model.Meta.table_name = table_name
model.Meta.region = region
if is_remote:
class DocumentTabModel(MapAttribute):
"""
Represents a document tab in the document model.
"""
document_tab_type = UnicodeAttribute(default="text")
document_tab_id = UnicodeAttribute()
document_tab_name = UnicodeAttribute()
document_tab_page = NumberAttribute(default=1)
document_tab_position_x = NumberAttribute()
document_tab_position_y = NumberAttribute()
document_tab_width = NumberAttribute(default=200)
document_tab_height = NumberAttribute(default=20)
document_tab_is_locked = BooleanAttribute(default=False)
document_tab_is_required = BooleanAttribute(default=True)
document_tab_anchor_string = UnicodeAttribute(default=None)
document_tab_anchor_ignore_if_not_present = BooleanAttribute(default=True)
document_tab_anchor_x_offset = NumberAttribute()
document_tab_anchor_y_offset = NumberAttribute()
class DocumentTab(model_interfaces.DocumentTab):
"""
ORM-agnostic wrapper for the DynamoDB DocumentTab model.
"""
def __init__(
self, # pylint: disable=too-many-arguments
document_tab_type=None,
document_tab_id=None,
"""Meta class for Project."""
table_name = "cla-{}-projects".format(stage)
if stage == "local":
host = "http://localhost:8000"
project_id = UnicodeAttribute(hash_key=True)
project_external_id = UnicodeAttribute()
project_name = UnicodeAttribute()
project_name_lower = UnicodeAttribute(null=True)
project_individual_documents = ListAttribute(of=DocumentModel, default=[])
project_corporate_documents = ListAttribute(of=DocumentModel, default=[])
project_member_documents = ListAttribute(of=DocumentModel, default=[])
project_icla_enabled = BooleanAttribute(default=True)
project_ccla_enabled = BooleanAttribute(default=True)
project_ccla_requires_icla_signature = BooleanAttribute(default=False)
project_external_id_index = ExternalProjectIndex()
project_acl = UnicodeSetAttribute(default=set())
class Project(model_interfaces.Project): # pylint: disable=too-many-public-methods
"""
ORM-agnostic wrapper for the DynamoDB Project model.
"""
def __init__(
self,
project_id=None,
project_external_id=None,
project_name=None,
project_name_lower=None,
project_icla_enabled=True,
from pynamodb.attributes import UnicodeAttribute, BooleanAttribute
app = Flask(__name__)
auth = HTTPBasicAuth()
class Task(Model):
class Meta:
table_name = os.environ.get('STAGE', 'dev') + '.tasks'
region = boto3.Session().region_name
host = 'http://localhost:8000' \
if not os.environ.get('LAMBDA_TASK_ROOT') else None
id = UnicodeAttribute(hash_key=True)
title = UnicodeAttribute()
description = UnicodeAttribute()
done = BooleanAttribute()
@auth.get_password
def get_password(username):
if username == 'miguel':
return 'python'
return None
@auth.error_handler
def unauthorized():
return make_response(jsonify({'error': 'Unauthorized access'}), 401)
@app.errorhandler(400)
def bad_request(error):
host = "http://localhost:8000"
write_capacity_units = int(cla.conf["DYNAMO_WRITE_UNITS"])
read_capacity_units = int(cla.conf["DYNAMO_READ_UNITS"])
signature_id = UnicodeAttribute(hash_key=True)
signature_external_id = UnicodeAttribute(null=True)
signature_project_id = UnicodeAttribute()
signature_document_minor_version = NumberAttribute()
signature_document_major_version = NumberAttribute()
signature_reference_id = UnicodeAttribute()
signature_reference_name = UnicodeAttribute(null=True)
signature_reference_name_lower = UnicodeAttribute(null=True)
signature_reference_type = UnicodeAttribute()
signature_type = UnicodeAttribute(default="cla")
signature_signed = BooleanAttribute(default=False)
signature_approved = BooleanAttribute(default=False)
signature_sign_url = UnicodeAttribute(null=True)
signature_return_url = UnicodeAttribute(null=True)
signature_callback_url = UnicodeAttribute(null=True)
signature_user_ccla_company_id = UnicodeAttribute(null=True)
signature_acl = UnicodeSetAttribute(default=set())
signature_project_index = ProjectSignatureIndex()
signature_reference_index = ReferenceSignatureIndex()
signature_envelope_id = UnicodeAttribute(null=True)
# Callback type refers to either Gerrit or GitHub
signature_return_url_type = UnicodeAttribute(null=True)
note = UnicodeAttribute(null=True)
signature_project_external_id = UnicodeAttribute(null=True)
signature_company_signatory_id = UnicodeAttribute(null=True)
signature_company_signatory_name = UnicodeAttribute(null=True)
signature_company_signatory_email = UnicodeAttribute(null=True)
signature_company_initial_manager_id = UnicodeAttribute(null=True)
table_name = "price_fairy_price_record"
region = 'ap-northeast-2'
product_id = UnicodeAttribute(hash_key=True)
crawled_at = UTCDateTimeAttribute(range_key=True)
item = MapAttribute()
class Product(Model):
class Meta:
table_name = "price_fairy_product"
region = 'ap-northeast-2'
id = UnicodeAttribute(hash_key=True)
do_crawl = BooleanAttribute(default=False)
created_at = UnicodeAttribute()
last_crawled_at = UTCDateTimeAttribute(null=True)
min_price = NumberAttribute(default=0) # min_price 가격 이상인 결과만 사용
queries = ListAttribute(of=ShoppingQuery)
lprice = NumberAttribute(null=True) # 현재까지 최저 가격
lprice_item = MapAttribute(null=True) # 최저가격의 item 정보
def update_lprice(self, lprice_item):
self.last_crawled_at = datetime.datetime.utcnow()
if lprice_item:
self.lprice_item = lprice_item
self.lprice = int(lprice_item['lprice'])