本文整理汇总了Python中zappa.zappa.Zappa类的典型用法代码示例。如果您正苦于以下问题:Python Zappa类的具体用法?Python Zappa怎么用?Python Zappa使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Zappa类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_load_credentials
def test_load_credentials(self):
z = Zappa()
z.aws_region = 'us-east-1'
z.load_credentials()
self.assertEqual(z.boto_session.region_name, 'us-east-1')
self.assertEqual(z.aws_region, 'us-east-1')
z.aws_region = 'eu-west-1'
z.load_credentials()
self.assertEqual(z.boto_session.region_name, 'eu-west-1')
self.assertEqual(z.aws_region, 'eu-west-1')
creds = {
'AWS_ACCESS_KEY_ID': 'AK123',
'AWS_SECRET_ACCESS_KEY': 'JKL456',
'AWS_DEFAULT_REGION': 'us-west-1'
}
with mock.patch.dict('os.environ', creds):
z.aws_region = None
z.load_credentials()
loaded_creds = z.boto_session._session.get_credentials()
self.assertEqual(loaded_creds.access_key, 'AK123')
self.assertEqual(loaded_creds.secret_key, 'JKL456')
self.assertEqual(z.boto_session.region_name, 'us-west-1')
开发者ID:hellking4u,项目名称:Zappa,代码行数:25,代码来源:tests.py
示例2: test_domain_name_match
def test_domain_name_match(self):
# Simple sanity check
zone = Zappa._get_best_match_zone(all_zones={ 'HostedZones': [
{
'Name': 'example.com.au.',
'Id': 'zone-correct'
}
]},
domain='www.example.com.au')
assert zone == 'zone-correct'
# No match test
zone = Zappa._get_best_match_zone(all_zones={'HostedZones': [
{
'Name': 'example.com.au.',
'Id': 'zone-incorrect'
}
]},
domain='something-else.com.au')
assert zone is None
# More involved, better match should win.
zone = Zappa._get_best_match_zone(all_zones={'HostedZones': [
{
'Name': 'example.com.au.',
'Id': 'zone-incorrect'
},
{
'Name': 'subdomain.example.com.au.',
'Id': 'zone-correct'
}
]},
domain='www.subdomain.example.com.au')
assert zone == 'zone-correct'
开发者ID:avengerpenguin,项目名称:Zappa,代码行数:34,代码来源:tests.py
示例3: handle
def handle(self, *args, **options):
if not options.has_key('environment'):
print("You must call rollback with an environment name. \n python manage.py rollback <environment> <revisions>")
return
from django.conf import settings
if not 'ZAPPA_SETTINGS' in dir(settings):
print("Please define your ZAPPA_SETTINGS in your settings file before deploying.")
return
zappa_settings = settings.ZAPPA_SETTINGS
# Set your configuration
project_name = settings.BASE_DIR.split(os.sep)[-1]
api_stage = options['environment'][0]
if api_stage not in zappa_settings.keys():
print("Please make sure that the environment '" + api_stage + "' is defined in your ZAPPA_SETTINGS in your settings file before deploying.")
return
lambda_name = project_name + '-' + api_stage
revision = options['revision'][0]
# Make your Zappa object
zappa = Zappa()
# Load your AWS credentials from ~/.aws/credentials
zappa.load_credentials()
print("Rolling back..")
response = zappa.rollback_lambda_function_version(lambda_name, versions_back=revision)
print("Done!")
开发者ID:codingjoe,项目名称:django-zappa,代码行数:31,代码来源:rollback.py
示例4: test_invoke_lambda_function
def test_invoke_lambda_function(self, session):
z = Zappa(session)
z.credentials_arn = 'arn:aws:iam::724336686645:role/ZappaLambdaExecution'
function_name = 'django-helloworld-unicode'
payload = '{"event": "hello"}'
response = z.invoke_lambda_function(function_name, payload)
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:7,代码来源:tests.py
示例5: test_rollback_lambda_function_version
def test_rollback_lambda_function_version(self, session):
z = Zappa(session)
z.credentials_arn = 'arn:aws:iam::724336686645:role/ZappaLambdaExecution'
function_name = 'django-helloworld-unicode'
too_many_versions = z.rollback_lambda_function_version(function_name, 99999)
self.assertFalse(too_many_versions)
function_arn = z.rollback_lambda_function_version(function_name, 1)
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:9,代码来源:tests.py
示例6: test_create_lambda_package
def test_create_lambda_package(self):
# mock the pip.get_installed_distributions() to include a package in lambda_packages so that the code
# for zipping pre-compiled packages gets called
mock_named_tuple = collections.namedtuple('mock_named_tuple', ['project_name'])
mock_return_val = [mock_named_tuple(lambda_packages.keys()[0])] # choose name of 1st package in lambda_packages
with mock.patch('pip.get_installed_distributions', return_value=mock_return_val):
z = Zappa()
path = z.create_lambda_zip(handler_file=os.path.realpath(__file__))
self.assertTrue(os.path.isfile(path))
os.remove(path)
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:10,代码来源:tests.py
示例7: handle
def handle(self, *args, **options):
"""
Execute the command.
"""
if not options.has_key('environment'):
print("You must call deploy with an environment name. \n python manage.py deploy <environment>")
return
from django.conf import settings
if not 'ZAPPA_SETTINGS' in dir(settings):
print("Please define your ZAPPA_SETTINGS in your settings file before deploying.")
return
zappa_settings = settings.ZAPPA_SETTINGS
# Set your configuration
project_name = settings.BASE_DIR.split(os.sep)[-1]
api_stage = options['environment'][0]
if api_stage not in zappa_settings.keys():
print("Please make sure that the environment '" + api_stage + "' is defined in your ZAPPA_SETTINGS in your settings file before deploying.")
return
lambda_name = project_name + '-' + api_stage
# Make your Zappa object
zappa = Zappa()
# Load your AWS credentials from ~/.aws/credentials
zappa.load_credentials()
try:
# Tail the available logs
all_logs = zappa.fetch_logs(lambda_name)
self.print_logs(all_logs)
# Keep polling, and print any new logs.
while True:
all_logs_again = zappa.fetch_logs(lambda_name)
new_logs = []
for log in all_logs_again:
if log not in all_logs:
new_logs.append(log)
self.print_logs(new_logs)
all_logs = all_logs + new_logs
except KeyboardInterrupt:
# Die gracefully
try:
sys.exit(0)
except SystemExit:
os._exit(0)
return
开发者ID:codingjoe,项目名称:django-zappa,代码行数:53,代码来源:tail.py
示例8: test_upload_remove_s3
def test_upload_remove_s3(self, session):
bucket_name = 'test_zappa_upload_s3'
z = Zappa(session)
zip_path = z.create_lambda_zip()
res = z.upload_to_s3(zip_path, bucket_name)
os.remove(zip_path)
self.assertTrue(res)
s3 = session.resource('s3')
# will throw ClientError with 404 if bucket doesn't exist
s3.meta.client.head_bucket(Bucket=bucket_name)
# will throw ClientError with 404 if object doesn't exist
s3.meta.client.head_object(
Bucket=bucket_name,
Key=zip_path,
)
res = z.remove_from_s3(zip_path, bucket_name)
self.assertTrue(res)
开发者ID:Doerge,项目名称:Zappa,代码行数:19,代码来源:tests.py
示例9: test_deploy_api_gateway
def test_deploy_api_gateway(self, session):
z = Zappa(session)
z.credentials_arn = 'arn:aws:iam::12345:role/ZappaLambdaExecution'
z.parameter_depth = 1
z.integration_response_codes = [200]
z.method_response_codes = [200]
z.http_methods = ['GET']
lambda_arn = 'arn:aws:lambda:us-east-1:12345:function:django-helloworld-unicode'
api_id = z.create_api_gateway_routes(lambda_arn)
endpoint_url = z.deploy_api_gateway(api_id, "test_stage")
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:12,代码来源:tests.py
示例10: handle
def handle(self, *args, **options):
"""
Execute the command.
"""
if not options.has_key('environment') or len(options['environment']) < 2:
print("You must call deploy with an environment name and command. \n python manage.py invoke <environment> <command>")
return
from django.conf import settings
if not 'ZAPPA_SETTINGS' in dir(settings):
print("Please define your ZAPPA_SETTINGS in your settings file before deploying.")
return
zappa_settings = settings.ZAPPA_SETTINGS
# Set your configuration
project_name = settings.BASE_DIR.split(os.sep)[-1]
api_stage = options['environment'][0]
if api_stage not in zappa_settings.keys():
print("Please make sure that the environment '" + api_stage + "' is defined in your ZAPPA_SETTINGS in your settings file before deploying.")
return
lambda_name = project_name + '-' + api_stage
# Make your Zappa object
zappa = Zappa()
# Invoke it!
command = {"command": ' '.join(options['environment'][1:])}
response = zappa.invoke_lambda_function(lambda_name, json.dumps(command), invocation_type='RequestResponse')
if response.has_key('LogResult'):
print(base64.b64decode(response['LogResult']))
else:
print(response)
import pdb
pdb.set_trace()
return
开发者ID:Doerge,项目名称:django-zappa,代码行数:41,代码来源:invoke.py
示例11: test_b64_pattern
def test_b64_pattern(self):
head = '<!DOCTYPE html>'
for code in ['400', '401', '402', '403', '404', '500']:
pattern = Zappa.selection_pattern(code)
document = base64.b64encode(head + code + random_string(50))
self.assertRegexpMatches(document, pattern)
for bad_code in ['200', '301', '302']:
document = base64.b64encode(head + bad_code + random_string(50))
self.assertNotRegexpMatches(document, pattern)
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:12,代码来源:tests.py
示例12: test_load_credentials
def test_load_credentials(self):
z = Zappa()
credentials = '[default]\naws_access_key_id = AK123\naws_secret_access_key = JKL456'
config = '[default]\noutput = json\nregion = us-east-1'
credentials_file = open('credentials','w')
credentials_file.write(credentials)
credentials_file.close()
config_file = open('config','w')
config_file.write(config)
config_file.close()
z.load_credentials('credentials', 'config')
os.remove('credentials')
os.remove('config')
self.assertTrue((z.access_key == "AK123"))
self.assertTrue((z.secret_key == "JKL456"))
self.assertTrue((z.aws_region == 'us-east-1'))
开发者ID:evelynmitchell,项目名称:Zappa,代码行数:22,代码来源:tests.py
示例13: test_create_api_gateway_routes
def test_create_api_gateway_routes(self, session):
z = Zappa(session)
z.parameter_depth = 1
z.integration_response_codes = [200]
z.method_response_codes = [200]
z.http_methods = ['GET']
z.credentials_arn = 'arn:aws:iam::12345:role/ZappaLambdaExecution'
lambda_arn = 'arn:aws:lambda:us-east-1:12345:function:helloworld'
with mock.patch('time.time', return_value=123.456):
api_id = z.create_api_gateway_routes(lambda_arn)
self.assertEqual(api_id, 'j27idab94h')
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:11,代码来源:tests.py
示例14: test_redirect_pattern
def test_redirect_pattern(self):
test_urls = [
# a regular endpoint url
'https://asdf1234.execute-api.us-east-1.amazonaws.com/env/path/to/thing',
# an external url (outside AWS)
'https://github.com/Miserlou/zappa/issues?q=is%3Aissue+is%3Aclosed',
# a local url
'/env/path/to/thing'
]
for code in ['301', '302']:
pattern = Zappa.selection_pattern(code)
for url in test_urls:
self.assertRegexpMatches(url, pattern)
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:15,代码来源:tests.py
示例15: test_create_lambda_function
def test_create_lambda_function(self, session):
bucket_name = 'lmbda'
zip_path = 'Spheres-dev-1454694878.zip'
z = Zappa(session)
z.aws_region = 'us-east-1'
z.load_credentials(session)
z.credentials_arn = 'arn:aws:iam::12345:role/ZappaLambdaExecution'
arn = z.create_lambda_function(
bucket=bucket_name,
s3_key=zip_path,
function_name='test_lmbda_function55',
handler='runme.lambda_handler'
)
arn = z.update_lambda_function(
bucket=bucket_name,
s3_key=zip_path,
function_name='test_lmbda_function55',
)
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:21,代码来源:tests.py
示例16: handle
def handle(self, *args, **options):
"""
Execute the command.
"""
if not options.has_key("environment"):
print("You must call deploy with an environment name. \n python manage.py deploy <environment>")
return
from django.conf import settings
if not "ZAPPA_SETTINGS" in dir(settings):
print("Please define your ZAPPA_SETTINGS in your settings file before deploying.")
return
zappa_settings = settings.ZAPPA_SETTINGS
# Set your configuration
project_name = settings.BASE_DIR.split(os.sep)[-1]
api_stage = options["environment"][0]
if api_stage not in zappa_settings.keys():
print(
"Please make sure that the environment '"
+ api_stage
+ "' is defined in your ZAPPA_SETTINGS in your settings file before deploying."
)
return
# Make your Zappa object
zappa = Zappa()
# Load environment-specific settings
s3_bucket_name = zappa_settings[api_stage]["s3_bucket"]
settings_file = zappa_settings[api_stage]["settings_file"]
if "~" in settings_file:
settings_file = settings_file.replace("~", os.path.expanduser("~"))
if not os.path.isfile(settings_file):
print("Please make sure your settings_file is properly defined.")
return
custom_settings = [
"http_methods",
"parameter_depth",
"integration_response_codes",
"method_response_codes",
"role_name",
"aws_region",
]
for setting in custom_settings:
if zappa_settings[api_stage].has_key(setting):
setattr(zappa, setting, zappa_settings[api_stage][setting])
# Load your AWS credentials from ~/.aws/credentials
zappa.load_credentials()
# Create the Lambda zip package (includes project and virtualenvironment)
# Also define the path the handler file so it can be copied to the zip root for Lambda.
current_file = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
handler_file = os.sep.join(current_file.split(os.sep)[0:-2]) + os.sep + "handler.py"
lambda_name = project_name + "-" + api_stage
zip_path = zappa.create_lambda_zip(lambda_name, handler_file=handler_file)
# Add this environment's Django settings to that zipfile
with open(settings_file, "r") as f:
contents = f.read()
all_contents = contents
if not zappa_settings[api_stage].has_key("domain"):
script_name = api_stage
else:
script_name = ""
if not "ZappaMiddleware" in all_contents:
print("\n\nWARNING!\n")
print("You do not have ZappaMiddleware in your remote settings's MIDDLEWARE_CLASSES.\n")
print("This means that some aspects of your application may not work!\n\n")
all_contents = all_contents + "\n# Automatically added by Zappa:\nSCRIPT_NAME='/" + script_name + "'\n"
f.close()
with open("zappa_settings.py", "w") as f:
f.write(all_contents)
with zipfile.ZipFile(zip_path, "a") as lambda_zip:
lambda_zip.write("zappa_settings.py", "zappa_settings.py")
lambda_zip.close()
os.unlink("zappa_settings.py")
# Upload it to S3
zip_arn = zappa.upload_to_s3(zip_path, s3_bucket_name)
# Register the Lambda function with that zip as the source
# You'll also need to define the path to your lambda_handler code.
lambda_arn = zappa.update_lambda_function(s3_bucket_name, zip_path, lambda_name)
# Finally, delete the local copy our zip package
if zappa_settings[api_stage].get("delete_zip", True):
os.remove(zip_path)
# Remove the uploaded zip from S3, because it is now registered..
#.........这里部分代码省略.........
开发者ID:collingreen,项目名称:django-zappa,代码行数:101,代码来源:update.py
示例17: test_fetch_logs
def test_fetch_logs(self, session):
z = Zappa(session)
z.credentials_arn = 'arn:aws:iam::12345:role/ZappaLambdaExecution'
events = z.fetch_logs('Spheres-demonstration')
self.assertTrue(events is not None)
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:5,代码来源:tests.py
示例18: test_get_api_url
def test_get_api_url(self, session):
z = Zappa(session)
z.credentials_arn = 'arn:aws:iam::724336686645:role/ZappaLambdaExecution'
url = z.get_api_url('Spheres-demonstration')
开发者ID:ConfidentCannabis,项目名称:Zappa,代码行数:4,代码来源:tests.py
示例19: ZappaCommand
class ZappaCommand(BaseCommand):
# Management command
can_import_settings = True
requires_system_checks = False
# Zappa settings
zappa = None
zappa_settings = None
api_stage = None
project_name = None
lambda_name = None
s3_bucket_name = None
settings_file = None
zip_path = None
vpc_config = None
memory_size = None
help = '''Deploy this project to AWS with Zappa.'''
def add_arguments(self, parser):
parser.add_argument('environment', nargs='+', type=str)
def __init__(self, *args, **kwargs):
super(ZappaCommand, self).__init__(*args, **kwargs)
self.zappa = Zappa()
def require_settings(self, args, options):
"""
Load the ZAPPA_SETTINGS as we expect it.
"""
if not options.has_key('environment'):
print(
"You must call deploy with an environment name. \n python manage.py deploy <environment>")
raise ImproperlyConfigured
from django.conf import settings
if not 'ZAPPA_SETTINGS' in dir(settings):
print(
"Please define your ZAPPA_SETTINGS in your settings file before deploying.")
raise ImproperlyConfigured
self.zappa_settings = settings.ZAPPA_SETTINGS
# Set your configuration
self.project_name = os.path.abspath(settings.BASE_DIR).split(os.sep)[-1]
if type(options['environment']) == list:
self.api_stage = options['environment'][0]
else:
self.api_stage = options['environment']
self.lambda_name = self.project_name + '-' + self.api_stage
if self.api_stage not in self.zappa_settings.keys():
print("Please make sure that the environment '" + self.api_stage +
"' is defined in your ZAPPA_SETTINGS in your settings file before deploying.")
raise ImproperlyConfigured
# Load environment-specific settings
self.s3_bucket_name = self.zappa_settings[self.api_stage]['s3_bucket']
self.vpc_config = self.zappa_settings[
self.api_stage].get('vpc_config', {})
self.memory_size = self.zappa_settings[
self.api_stage].get('memory_size', 512)
self.settings_file = self.zappa_settings[
self.api_stage]['settings_file']
if '~' in self.settings_file:
self.settings_file = self.settings_file.replace(
'~', os.path.expanduser('~'))
if not os.path.isfile(self.settings_file):
print("Please make sure your settings_file is properly defined.")
raise ImproperlyConfigured
custom_settings = [
'http_methods',
'parameter_depth',
'integration_response_codes',
'method_response_codes',
'role_name',
'aws_region'
]
for setting in custom_settings:
if self.zappa_settings[self.api_stage].has_key(setting):
setattr(self.zappa, setting, self.zappa_settings[
self.api_stage][setting])
def create_package(self):
"""
Ensure that the package can be properly configured,
and then create it.
"""
# Create the Lambda zip package (includes project and virtualenvironment)
# Also define the path the handler file so it can be copied to the zip
# root for Lambda.
current_file = os.path.dirname(os.path.abspath(
inspect.getfile(inspect.currentframe())))
handler_file = os.sep.join(current_file.split(os.sep)[
0:-2]) + os.sep + 'handler.py'
#.........这里部分代码省略.........
开发者ID:aberdean,项目名称:django-zappa,代码行数:101,代码来源:zappa_command.py
示例20: handle
def handle(self, *args, **options):
"""
Execute the command.
"""
if not options.has_key('environment'):
print("You must call deploy with an environment name. \n python manage.py deploy <environment>")
return
from django.conf import settings
if not 'ZAPPA_SETTINGS' in dir(settings):
print("Please define your ZAPPA_SETTINGS in your settings file before deploying.")
return
zappa_settings = settings.ZAPPA_SETTINGS
# Set your configuration
project_name = settings.BASE_DIR.split(os.sep)[-1]
api_stage = options['environment'][0]
if api_stage not in zappa_settings.keys():
print("Please make sure that the environment '" + api_stage + "' is defined in your ZAPPA_SETTINGS in your settings file before deploying.")
return
# Make your Zappa object
zappa = Zappa()
# Load environment-specific settings
s3_bucket_name = zappa_settings[api_stage]['s3_bucket']
settings_file = zappa_settings[api_stage]['settings_file']
if '~' in settings_file:
settings_file = settings_file.replace('~', os.path.expanduser('~'))
if not os.path.isfile(settings_file):
print("Please make sure your settings_file is properly defined.")
return
custom_settings = [
'http_methods',
'parameter_depth',
'integration_response_codes',
'method_response_codes',
'role_name',
'aws_region'
]
for setting in custom_settings:
if zappa_settings[api_stage].has_key(setting):
setattr(zappa, setting, zappa_settings[api_stage][setting])
# Load your AWS credentials from ~/.aws/credentials
zappa.load_credentials()
# Create the Lambda zip package (includes project and virtualenvironment)
# Also define the path the handler file so it can be copied to the zip root for Lambda.
current_file = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
handler_file = os.sep.join(current_file.split(os.sep)[0:-2]) + os.sep + 'handler.py'
lambda_name = project_name + '-' + api_stage
zip_path = zappa.create_lambda_zip(lambda_name, handler_file=handler_file)
# Add this environment's Django settings to that zipfile
with zipfile.ZipFile(zip_path, 'a') as lambda_zip:
lambda_zip.write(settings_file, 'zappa_settings.py')
lambda_zip.close()
# Upload it to S3
zip_arn = zappa.upload_to_s3(zip_path, s3_bucket_name)
# Register the Lambda function with that zip as the source
# You'll also need to define the path to your lambda_handler code.
lambda_arn = zappa.update_lambda_function(s3_bucket_name, zip_path, lambda_name)
# Get the URL!
endpoint_url = zappa.get_api_url(lambda_name)
# Finally, delete the local copy our zip package
os.remove(zip_path)
# Remove the uploaded zip from S3, because it is now registered..
zappa.remove_from_s3(zip_path, s3_bucket_name)
print("Your updated Zappa deployment is live!")
return
开发者ID:davinirjr,项目名称:django-zappa,代码行数:81,代码来源:update.py
注:本文中的zappa.zappa.Zappa类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论