Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
name = UnicodeAttribute(hash_key=True)
date_created = UTCDateTimeAttribute(default=datetime.utcnow)
class Location(MapAttribute):
lat = NumberAttribute(attr_name='latitude')
lng = NumberAttribute(attr_name='longitude')
name = UnicodeAttribute()
class Person(MapAttribute):
fname = UnicodeAttribute(attr_name='firstName')
lname = UnicodeAttribute(null=True)
age = NumberAttribute(null=True)
is_male = BooleanAttribute(attr_name='is_dude')
def foo(self):
return 1
class ComplexModel(Model):
class Meta:
table_name = 'ComplexModel'
person = Person(attr_name='weird_person')
key = NumberAttribute(hash_key=True)
class OfficeEmployee(Model):
class Meta:
return ret
class GerritModel(BaseModel):
"""
Represents a Gerrit Instance in the database.
"""
class Meta:
"""Meta class for User."""
table_name = 'cla-{}-gerrit-instances'.format(stage)
if stage == 'local':
host = 'http://localhost:8000'
gerrit_id = UnicodeAttribute(hash_key=True)
project_id = UnicodeAttribute()
gerrit_name = UnicodeAttribute()
gerrit_url = UnicodeAttribute()
group_id_icla = UnicodeAttribute(null=True)
group_id_ccla = UnicodeAttribute(null=True)
group_name_icla = UnicodeAttribute(null=True)
group_name_ccla = UnicodeAttribute(null=True)
class Gerrit(model_interfaces.Gerrit): # pylint: disable=too-many-public-methods
"""
ORM-agnostic wrapper for the DynamoDB Gerrit model.
"""
def __init__(self, gerrit_id=None, gerrit_name=None,
project_id=None, gerrit_url=None, group_id_icla=None, group_id_ccla=None):
super(Gerrit).__init__()
self.model = GerritModel()
self.model.gerrit_id = gerrit_id
self.model.gerrit_name = gerrit_name
self.model.project_id = project_id
self.model.gerrit_url = gerrit_url
from historical.models import (
AWSHistoricalMixin,
CurrentHistoricalModel,
DurableHistoricalModel,
HistoricalPollingBaseModel,
HistoricalPollingEventDetail,
)
VERSION = 1
class VPCModel:
"""VPC specific fields for DynamoDB."""
VpcId = UnicodeAttribute()
State = UnicodeAttribute()
CidrBlock = UnicodeAttribute()
IsDefault = BooleanAttribute()
Name = UnicodeAttribute(null=True)
Region = UnicodeAttribute()
class DurableVPCModel(DurableHistoricalModel, AWSHistoricalMixin, VPCModel):
"""The Durable Table model for VPC."""
class Meta:
"""Table details"""
table_name = 'HistoricalVPCDurableTable'
region = CURRENT_REGION
tech = 'vpc'
from pynamodb.models import Model
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)
class ExternalProjectIndex(GlobalSecondaryIndex):
"""
This class represents a global secondary index for querying projects by external ID.
"""
class Meta:
"""Meta class for external ID project index."""
index_name = "external-project-index"
write_capacity_units = int(cla.conf["DYNAMO_WRITE_UNITS"])
read_capacity_units = int(cla.conf["DYNAMO_READ_UNITS"])
# All attributes are projected - not sure if this is necessary.
projection = AllProjection()
# This attribute is the hash key for the index.
project_external_id = UnicodeAttribute(hash_key=True)
class ExternalCompanyIndex(GlobalSecondaryIndex):
"""
This class represents a global secondary index for querying companies by external ID.
"""
class Meta:
"""Meta class for external ID company index."""
index_name = "external-company-index"
write_capacity_units = int(cla.conf["DYNAMO_WRITE_UNITS"])
read_capacity_units = int(cla.conf["DYNAMO_READ_UNITS"])
# All attributes are projected - not sure if this is necessary.
projection = AllProjection()
"""Meta class for Signature."""
table_name = "cla-{}-signatures".format(stage)
if stage == "local":
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)
class GitHubOrgModel(BaseModel):
"""
Represents a Github Organization in the database.
Company_id, project_id are deprecated now that organizations are under an SFDC ID.
"""
class Meta:
"""Meta class for User."""
table_name = 'cla-{}-github-orgs'.format(stage)
if stage == 'local':
host = 'http://localhost:8000'
organization_name = UnicodeAttribute(hash_key=True)
organization_installation_id = NumberAttribute(null=True)
organization_sfid = UnicodeAttribute()
organization_sfid_index = GithubOrgSFIndex()
organization_project_id = UnicodeAttribute(null=True)
organization_company_id = UnicodeAttribute(null=True)
class GitHubOrg(model_interfaces.GitHubOrg): # pylint: disable=too-many-public-methods
"""
ORM-agnostic wrapper for the DynamoDB GitHubOrg model.
"""
def __init__(self, organization_name=None, organization_installation_id=None, organization_sfid=None):
super(GitHubOrg).__init__()
self.model = GitHubOrgModel()
self.model.organization_name = organization_name
self.model.organization_installation_id = organization_installation_id
self.model.organization_sfid = organization_sfid
def to_dict(self):
class GitHubOrgModel(BaseModel):
"""
Represents a Github Organization in the database.
Company_id, project_id are deprecated now that organizations are under an SFDC ID.
"""
class Meta:
"""Meta class for User."""
table_name = 'cla-{}-github-orgs'.format(stage)
if stage == 'local':
host = 'http://localhost:8000'
organization_name = UnicodeAttribute(hash_key=True)
organization_installation_id = NumberAttribute(null=True)
organization_sfid = UnicodeAttribute()
organization_sfid_index = GithubOrgSFIndex()
organization_project_id = UnicodeAttribute(null=True)
organization_company_id = UnicodeAttribute(null=True)
class GitHubOrg(model_interfaces.GitHubOrg): # pylint: disable=too-many-public-methods
"""
ORM-agnostic wrapper for the DynamoDB GitHubOrg model.
"""
def __init__(self, organization_name=None, organization_installation_id=None, organization_sfid=None):
super(GitHubOrg).__init__()
self.model = GitHubOrgModel()
self.model.organization_name = organization_name
self.model.organization_installation_id = organization_installation_id
self.model.organization_sfid = organization_sfid
def to_dict(self):
ret = dict(self.model)
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)
signature_company_initial_manager_name = UnicodeAttribute(null=True)
signature_company_initial_manager_email = UnicodeAttribute(null=True)
signature_company_secondary_manager_list = JSONAttribute(null=True)
signature_company_signatory_index = SignatureCompanySignatoryIndex()
signature_company_initial_manager_index = SignatureCompanyInitialManagerIndex()
project_signature_external_id_index = SignatureProjectExternalIndex()
# whitelists are only used by CCLAs
domain_whitelist = ListAttribute(null=True)
email_whitelist = ListAttribute(null=True)
github_whitelist = ListAttribute(null=True)
github_org_whitelist = ListAttribute(null=True)
class Signature(model_interfaces.Signature): # pylint: disable=too-many-public-methods
"""
projection = AllProjection()
email = UnicodeAttribute(hash_key=True)
class User(Model):
class Meta:
table_name = "users"
host = "http://localhost:8000"
def __init__(self, hash_key=None, range_key=None, **args):
Model.__init__(self, hash_key, range_key, **args)
if not self.id:
self.id = str(uuid.uuid4())
id = UnicodeAttribute(hash_key=True)
email = UnicodeAttribute(null=False)
email_index = UserEmailIndex()
first_name = UnicodeAttribute(null=False)
last_name = UnicodeAttribute(null=False)
password = PasswordAttribute(null=False)
def check_password(self, password):
return check_password_hash(self.password, password)
if not User.exists():
User.create_table(read_capacity_units=1, write_capacity_units=1, wait=True)