astronomer/airflow/version_check/models.py
def ensure_singleton(cls):
        """
        Ensure that the singleton row exists in this table
        """
        with create_session() as session:
            # To keep PG logs quieter (it shows an ERROR for the PK violation),
            # we try and select first
            if session.query(cls).get({"singleton": True}) is not None:
                return

            try:
                session.bulk_save_objects([cls(singleton=True)])
            except sqlalchemy.exc.IntegrityError:
                # Already exists, we're good
                session.rollback()

        Ensure that the singleton row exists in this table
        
Ensure that the singleton row exists in this table
Ensure that the singleton row exists in this table
def ensure_singleton(cls):
        with create_session() as session:
            if session.query(cls).get({"singleton": True}) is not None:
                return
            try:
                session.bulk_save_objects([cls(singleton=True)])
            except sqlalchemy.exc.IntegrityError:
                session.rollback()
def ensure_singleton ( cls ) : with create_session ( ) as session : if session . query ( cls ) . get ( { "singleton" : True } ) is not None : return try : session . bulk_save_objects ( [ cls ( singleton = True ) ] ) except sqlalchemy . exc . IntegrityError : session . rollback ( )
Ensure that the singleton row exists in this table
Ensure that the singleton row exists in this table
"""
        Ensure that the singleton row exists in this table
        """ # To keep PG logs quieter (it shows an ERROR for the PK violation), # we try and select first # Already exists, we're good
cls