def create()

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

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

def create(**kwargs):
    """
    Creates a new certificate.
    """
    if "destinations" in kwargs:
        validate_no_duplicate_destinations(kwargs["destinations"])

    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.get("common_name"),
            "san": ",".join(
                str(x.value) for x in kwargs["extensions"]["sub_alt_names"]["names"]
            ),
        }
        current_app.logger.error(log_data, exc_info=True)
        capture_exception()
        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:
        # ACME path
        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),
        )
        log_data = {
            "function": "lemur.certificates.service.create",
            "owner": cert.owner,
            "name": cert.name,
            "serial": cert.serial,
            "issuer": cert.issuer,
            "not_after": cert.not_after.format('YYYY-MM-DD HH:mm:ss'),
            "not_before": cert.not_before.format('YYYY-MM-DD HH:mm:ss'),
            "sans": str(', '.join([domain.name for domain in cert.domains])),
        }
        current_app.logger.info(log_data)

    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

        if not current_app.config.get("ACME_DISABLE_AUTORESOLVE", False):
            fetch_acme_cert.apply_async((pending_cert.id, kwargs.get("async_reissue_notification_cert_id", None)), countdown=5)

    return cert