def handle(self, **kwargs):
self.write('Bootstrapping Promgen')
if not os.path.exists(settings.CONFIG_DIR):
self.write('Creating config directory {} ', settings.CONFIG_DIR)
os.makedirs(settings.CONFIG_DIR)
if not os.path.exists(settings.PROMGEN_CONFIG):
path = os.path.join(settings.BASE_DIR, 'promgen', 'tests', 'examples', 'promgen.yml')
self.write('Creating promgen config {} from {}', settings.PROMGEN_CONFIG, path)
shutil.copy(path, settings.PROMGEN_CONFIG)
self.write_setting('SECRET_KEY', default=settings.SECRET_KEY)
self.write_setting('DATABASE_URL', test=dj_database_url.parse)
# Schemes based on list of supported brokers
# http://docs.celeryproject.org/en/latest/getting-started/brokers/index.html
self.write_setting('CELERY_BROKER_URL', test=URLValidator(schemes=['redis', 'amqp', 'sqs']))
python类parse()的实例源码
def load(url, verbose=False):
dj_details = dj_database_url.parse(url)
database_cls = DJANGO_ENGINE_TO_DBCONN_MAP.get(dj_details['ENGINE'])
if database_cls is None:
raise DatabaseUrlError(
'Unable to determine the database from the URL')
return database_cls.create_from_django_database(dj_details, verbose)
def handle(self, db, **kwargs):
settings.DATABASES['promgen'] = dj_database_url.parse(db)
connections = ConnectionHandler(settings.DATABASES)
with connections['promgen'].cursor() as c:
services = {}
c.execute('SELECT * FROM service')
for row in dictfetchall(c):
services[row['id']], _ = models.Service.objects.get_or_create(
name=row['name']
)
projects = {}
c.execute('SELECT * FROM project')
for row in dictfetchall(c):
projects[row['id']], _ = models.Project.objects.get_or_create(
name=row['name'],
service_id=services[row['service_id']].id
)
if row['mail_address']:
models.Sender.get_or_create(
obj=projects[row['id']],
sender='promgen.notification.email',
value=row['mail_address'],
)
if row['hipchat_channel']:
models.Sender.get_or_create(
obj=projects[row['id']],
sender='promgen.notification.ikasan',
value=row['hipchat_channel'],
)
if row['line_notify_access_token']:
models.Sender.get_or_create(
obj=projects[row['id']],
sender='promgen.notification.linenotify',
value=row['line_notify_access_token'],
password=True,
)
c.execute('SELECT * FROM rule')
for row in dictfetchall(c):
models.Rule.objects.update_or_create(
name=row['alert_clause'],
service=services[row['service_id']],
defaults={
'duration': row['for_clause'],
'clause': row['if_clause'],
'labels': convert_to_json(row['labels_clause'].strip()),
'annotations': convert_to_json(row['annotations_clause'].strip()),
}
)