airflow/api/common/experimental/pool.py
def get_pool(name, session=None):
    """Get pool by a given name."""
    if not (name and name.strip()):
        raise AirflowBadRequest("Pool name shouldn't be empty")

    pool = session.query(Pool).filter_by(pool=name).first()
    if pool is None:
        raise PoolNotFound("Pool '%s' doesn't exist" % name)

    return pool
def get_pool(name, session=None):
    """Get pool by a given name."""
    if not (name and name.strip()):
        raise AirflowBadRequest("Pool name shouldn't be empty")

    pool = session.query(Pool).filter_by(pool=name).first()
    if pool is None:
        raise PoolNotFound("Pool '%s' doesn't exist" % name)

    return pool
def get_pool ( name , session = None ) : if not ( name and name . strip ( ) ) : raise AirflowBadRequest ( "Pool name shouldn't be empty" ) pool = session . query ( Pool ) . filter_by ( pool = name ) . first ( ) if pool is None : raise PoolNotFound ( "Pool '%s' doesn't exist" % name ) return pool
Get pool by a given name.
Get pool by a given name .