python类parse()的实例源码

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']))

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()),
                    }
                )