def create()

in lemur/certificates/service.py [0:0]

• 47 lines of code
• 10 McCabe index (conditional complexity)

def create(**kwargs):
    """
    Creates a new certificate.
    """
    try:
        cert_body, private_key, cert_chain, external_id, csr = mint(**kwargs)
    except Exception:
        log_data = {
            "message": "Exception minting certificate",
            "issuer": kwargs["authority"].name,
            "cn": kwargs["common_name"],
        }
        current_app.logger.error(log_data, exc_info=True)
        sentry.captureException()
        raise
    kwargs["body"] = cert_body
    kwargs["private_key"] = private_key
    kwargs["chain"] = cert_chain
    kwargs["external_id"] = external_id
    kwargs["csr"] = csr

    roles = create_certificate_roles(**kwargs)

    if kwargs.get("roles"):
        kwargs["roles"] += roles
    else:
        kwargs["roles"] = roles

    if cert_body:
        cert = Certificate(**kwargs)
        kwargs["creator"].certificates.append(cert)
    else:
        cert = PendingCertificate(**kwargs)
        kwargs["creator"].pending_certificates.append(cert)

    cert.authority = kwargs["authority"]

    database.commit()

    if isinstance(cert, Certificate):
        certificate_issued.send(certificate=cert, authority=cert.authority)
        metrics.send(
            "certificate_issued",
            "counter",
            1,
            metric_tags=dict(owner=cert.owner, issuer=cert.issuer),
        )

    if isinstance(cert, PendingCertificate):
        # We need to refresh the pending certificate to avoid "Instance is not bound to a Session; "
        # "attribute refresh operation cannot proceed"
        pending_cert = database.session_query(PendingCertificate).get(cert.id)
        from lemur.common.celery import fetch_acme_cert, fetch_cert

        acme = database.get(Authority, cert.authority_id).plugin_name == "acme-issuer"
        if not acme and not current_app.config.get("DISABLE_AUTORESOLVE_NON_ACME", False):
            fetch_cert.apply_async((pending_cert.id,), countdown=5)
        if acme and not current_app.config.get("ACME_DISABLE_AUTORESOLVE", False):
            fetch_acme_cert.apply_async((pending_cert.id,), countdown=5)

    return cert