We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Tasks in an infinite slots pool are never scheduled
**Apache Airflow version**: v2.0.0 and up
**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): not tested with K8
**Environment**:
all

**What happened**:

Executing the unit test included below, or create an infinite pool ( `-1` slots ) and tasks that should be executed in that pool.
```
INFO     airflow.jobs.scheduler_job.SchedulerJob:scheduler_job.py:991 Not scheduling since there are -1 open slots in pool test_scheduler_verify_infinite_pool
```

**What you expected to happen**:

To schedule tasks, or to drop support for infinite slots pools?

**How to reproduce it**:
easiest one is this unit test:
```
def test_scheduler_verify_infinite_pool(self):
    """
    Test that TIs are still scheduled if we only have one infinite pool.
    """
    dag = DAG(dag_id='test_scheduler_verify_infinite_pool', start_date=DEFAULT_DATE)
    BashOperator(
        task_id='test_scheduler_verify_infinite_pool_t0',
        dag=dag,
        owner='airflow',
        pool='test_scheduler_verify_infinite_pool',
        bash_command='echo hi',
    )

    dagbag = DagBag(
        dag_folder=os.path.join(settings.DAGS_FOLDER, "no_dags.py"),
        include_examples=False,
        read_dags_from_db=True,
    )
    dagbag.bag_dag(dag=dag, root_dag=dag)
    dagbag.sync_to_db()

    session = settings.Session()
    pool = Pool(pool='test_scheduler_verify_infinite_pool', slots=-1)
    session.add(pool)
    session.commit()

    dag = SerializedDAG.from_dict(SerializedDAG.to_dict(dag))

    scheduler = SchedulerJob(executor=self.null_exec)
    scheduler.processor_agent = mock.MagicMock()

    dr = dag.create_dagrun(
        run_type=DagRunType.SCHEDULED,
        execution_date=DEFAULT_DATE,
        state=State.RUNNING,
    )
    scheduler._schedule_dag_run(dr, {}, session)

    task_instances_list = scheduler._executable_task_instances_to_queued(max_tis=32, session=session)

    # Let's make sure we don't end up with a `max_tis` == 0
    assert len(task_instances_list) >= 1
```

**Anything else we need to know**:

Overall I'm not sure whether it's worth fixing in those various spots:
https://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L908
https://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L971
https://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L988
https://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L1041
https://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L1056

Or whether to restrict `-1` ( infinite ) slots in pools:
https://github.com/bperson/airflow/blob/master/airflow/models/pool.py#L49

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/models/pool.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 
19 from typing import Dict, Iterable, Optional, Tuple
20 
21 from sqlalchemy import Column, Integer, String, Text, func
22 from sqlalchemy.orm.session import Session
23 
24 from airflow.exceptions import AirflowException
25 from airflow.models.base import Base
26 from airflow.ti_deps.dependencies_states import EXECUTION_STATES
27 from airflow.typing_compat import TypedDict
28 from airflow.utils.session import provide_session
29 from airflow.utils.sqlalchemy import nowait, with_row_locks
30 from airflow.utils.state import State
31 
32 
33 class PoolStats(TypedDict):
34     """Dictionary containing Pool Stats"""
35 
36     total: int
37     running: int
38     queued: int
39     open: int
40 
41 
42 class Pool(Base):
43     """the class to get Pool info."""
44 
45     __tablename__ = "slot_pool"
46 
47     id = Column(Integer, primary_key=True)
48     pool = Column(String(256), unique=True)
49     # -1 for infinite
50     slots = Column(Integer, default=0)
51     description = Column(Text)
52 
53     DEFAULT_POOL_NAME = 'default_pool'
54 
55     def __repr__(self):
56         return str(self.pool)  # pylint: disable=E0012
57 
58     @staticmethod
59     @provide_session
60     def get_pool(pool_name, session: Session = None):
61         """
62         Get the Pool with specific pool name from the Pools.
63 
64         :param pool_name: The pool name of the Pool to get.
65         :param session: SQLAlchemy ORM Session
66         :return: the pool object
67         """
68         return session.query(Pool).filter(Pool.pool == pool_name).first()
69 
70     @staticmethod
71     @provide_session
72     def get_default_pool(session: Session = None):
73         """
74         Get the Pool of the default_pool from the Pools.
75 
76         :param session: SQLAlchemy ORM Session
77         :return: the pool object
78         """
79         return Pool.get_pool(Pool.DEFAULT_POOL_NAME, session=session)
80 
81     @staticmethod
82     @provide_session
83     def slots_stats(
84         *,
85         lock_rows: bool = False,
86         session: Session = None,
87     ) -> Dict[str, PoolStats]:
88         """
89         Get Pool stats (Number of Running, Queued, Open & Total tasks)
90 
91         If ``lock_rows`` is True, and the database engine in use supports the ``NOWAIT`` syntax, then a
92         non-blocking lock will be attempted -- if the lock is not available then SQLAlchemy will throw an
93         OperationalError.
94 
95         :param lock_rows: Should we attempt to obtain a row-level lock on all the Pool rows returns
96         :param session: SQLAlchemy ORM Session
97         """
98         from airflow.models.taskinstance import TaskInstance  # Avoid circular import
99 
100         pools: Dict[str, PoolStats] = {}
101 
102         query = session.query(Pool.pool, Pool.slots)
103 
104         if lock_rows:
105             query = with_row_locks(query, session=session, **nowait(session))
106 
107         pool_rows: Iterable[Tuple[str, int]] = query.all()
108         for (pool_name, total_slots) in pool_rows:
109             pools[pool_name] = PoolStats(total=total_slots, running=0, queued=0, open=0)
110 
111         state_count_by_pool = (
112             session.query(TaskInstance.pool, TaskInstance.state, func.sum(TaskInstance.pool_slots))
113             .filter(TaskInstance.state.in_(list(EXECUTION_STATES)))
114             .group_by(TaskInstance.pool, TaskInstance.state)
115         ).all()
116 
117         # calculate queued and running metrics
118         count: int
119         for (pool_name, state, count) in state_count_by_pool:
120             stats_dict: Optional[PoolStats] = pools.get(pool_name)
121             if not stats_dict:
122                 continue
123             # TypedDict key must be a string literal, so we use if-statements to set value
124             if state == "running":
125                 stats_dict["running"] = count
126             elif state == "queued":
127                 stats_dict["queued"] = count
128             else:
129                 raise AirflowException(f"Unexpected state. Expected values: {EXECUTION_STATES}.")
130 
131         # calculate open metric
132         for pool_name, stats_dict in pools.items():
133             if stats_dict["total"] == -1:
134                 # -1 means infinite
135                 stats_dict["open"] = -1
136             else:
137                 stats_dict["open"] = stats_dict["total"] - stats_dict["running"] - stats_dict["queued"]
138 
139         return pools
140 
141     def to_json(self):
142         """
143         Get the Pool in a json structure
144 
145         :return: the pool object in json format
146         """
147         return {
148             'id': self.id,
149             'pool': self.pool,
150             'slots': self.slots,
151             'description': self.description,
152         }
153 
154     @provide_session
155     def occupied_slots(self, session: Session):
156         """
157         Get the number of slots used by running/queued tasks at the moment.
158 
159         :param session: SQLAlchemy ORM Session
160         :return: the used number of slots
161         """
162         from airflow.models.taskinstance import TaskInstance  # Avoid circular import
163 
164         return int(
165             session.query(func.sum(TaskInstance.pool_slots))
166             .filter(TaskInstance.pool == self.pool)
167             .filter(TaskInstance.state.in_(list(EXECUTION_STATES)))
168             .scalar()
169             or 0
170         )
171 
172     @provide_session
173     def running_slots(self, session: Session):
174         """
175         Get the number of slots used by running tasks at the moment.
176 
177         :param session: SQLAlchemy ORM Session
178         :return: the used number of slots
179         """
180         from airflow.models.taskinstance import TaskInstance  # Avoid circular import
181 
182         return int(
183             session.query(func.sum(TaskInstance.pool_slots))
184             .filter(TaskInstance.pool == self.pool)
185             .filter(TaskInstance.state == State.RUNNING)
186             .scalar()
187             or 0
188         )
189 
190     @provide_session
191     def queued_slots(self, session: Session):
192         """
193         Get the number of slots used by queued tasks at the moment.
194 
195         :param session: SQLAlchemy ORM Session
196         :return: the used number of slots
197         """
198         from airflow.models.taskinstance import TaskInstance  # Avoid circular import
199 
200         return int(
201             session.query(func.sum(TaskInstance.pool_slots))
202             .filter(TaskInstance.pool == self.pool)
203             .filter(TaskInstance.state == State.QUEUED)
204             .scalar()
205             or 0
206         )
207 
208     @provide_session
209     def open_slots(self, session: Session) -> float:
210         """
211         Get the number of slots open at the moment.
212 
213         :param session: SQLAlchemy ORM Session
214         :return: the number of slots
215         """
216         if self.slots == -1:
217             return float('inf')
218         else:
219             return self.slots - self.occupied_slots(session)
220
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/models/pool.py b/airflow/models/pool.py
--- a/airflow/models/pool.py
+++ b/airflow/models/pool.py
@@ -106,6 +106,8 @@
 
         pool_rows: Iterable[Tuple[str, int]] = query.all()
         for (pool_name, total_slots) in pool_rows:
+            if total_slots == -1:
+                total_slots = float('inf')  # type: ignore
             pools[pool_name] = PoolStats(total=total_slots, running=0, queued=0, open=0)
 
         state_count_by_pool = (
@@ -115,8 +117,10 @@
         ).all()
 
         # calculate queued and running metrics
-        count: int
         for (pool_name, state, count) in state_count_by_pool:
+            # Some databases return decimal.Decimal here.
+            count = int(count)
+
             stats_dict: Optional[PoolStats] = pools.get(pool_name)
             if not stats_dict:
                 continue

{"golden_diff": "diff --git a/airflow/models/pool.py b/airflow/models/pool.py\n--- a/airflow/models/pool.py\n+++ b/airflow/models/pool.py\n@@ -106,6 +106,8 @@\n \n         pool_rows: Iterable[Tuple[str, int]] = query.all()\n         for (pool_name, total_slots) in pool_rows:\n+            if total_slots == -1:\n+                total_slots = float('inf')  # type: ignore\n             pools[pool_name] = PoolStats(total=total_slots, running=0, queued=0, open=0)\n \n         state_count_by_pool = (\n@@ -115,8 +117,10 @@\n         ).all()\n \n         # calculate queued and running metrics\n-        count: int\n         for (pool_name, state, count) in state_count_by_pool:\n+            # Some databases return decimal.Decimal here.\n+            count = int(count)\n+\n             stats_dict: Optional[PoolStats] = pools.get(pool_name)\n             if not stats_dict:\n                 continue\n", "issue": "Tasks in an infinite slots pool are never scheduled\n**Apache Airflow version**: v2.0.0 and up\r\n**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): not tested with K8\r\n**Environment**:\r\nall\r\n\r\n**What happened**:\r\n\r\nExecuting the unit test included below, or create an infinite pool ( `-1` slots ) and tasks that should be executed in that pool.\r\n```\r\nINFO     airflow.jobs.scheduler_job.SchedulerJob:scheduler_job.py:991 Not scheduling since there are -1 open slots in pool test_scheduler_verify_infinite_pool\r\n```\r\n\r\n**What you expected to happen**:\r\n\r\nTo schedule tasks, or to drop support for infinite slots pools?\r\n\r\n**How to reproduce it**:\r\neasiest one is this unit test:\r\n```\r\ndef test_scheduler_verify_infinite_pool(self):\r\n    \"\"\"\r\n    Test that TIs are still scheduled if we only have one infinite pool.\r\n    \"\"\"\r\n    dag = DAG(dag_id='test_scheduler_verify_infinite_pool', start_date=DEFAULT_DATE)\r\n    BashOperator(\r\n        task_id='test_scheduler_verify_infinite_pool_t0',\r\n        dag=dag,\r\n        owner='airflow',\r\n        pool='test_scheduler_verify_infinite_pool',\r\n        bash_command='echo hi',\r\n    )\r\n\r\n    dagbag = DagBag(\r\n        dag_folder=os.path.join(settings.DAGS_FOLDER, \"no_dags.py\"),\r\n        include_examples=False,\r\n        read_dags_from_db=True,\r\n    )\r\n    dagbag.bag_dag(dag=dag, root_dag=dag)\r\n    dagbag.sync_to_db()\r\n\r\n    session = settings.Session()\r\n    pool = Pool(pool='test_scheduler_verify_infinite_pool', slots=-1)\r\n    session.add(pool)\r\n    session.commit()\r\n\r\n    dag = SerializedDAG.from_dict(SerializedDAG.to_dict(dag))\r\n\r\n    scheduler = SchedulerJob(executor=self.null_exec)\r\n    scheduler.processor_agent = mock.MagicMock()\r\n\r\n    dr = dag.create_dagrun(\r\n        run_type=DagRunType.SCHEDULED,\r\n        execution_date=DEFAULT_DATE,\r\n        state=State.RUNNING,\r\n    )\r\n    scheduler._schedule_dag_run(dr, {}, session)\r\n\r\n    task_instances_list = scheduler._executable_task_instances_to_queued(max_tis=32, session=session)\r\n\r\n    # Let's make sure we don't end up with a `max_tis` == 0\r\n    assert len(task_instances_list) >= 1\r\n```\r\n\r\n**Anything else we need to know**:\r\n\r\nOverall I'm not sure whether it's worth fixing in those various spots:\r\nhttps://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L908\r\nhttps://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L971\r\nhttps://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L988\r\nhttps://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L1041\r\nhttps://github.com/bperson/airflow/blob/master/airflow/jobs/scheduler_job.py#L1056\r\n\r\nOr whether to restrict `-1` ( infinite ) slots in pools:\r\nhttps://github.com/bperson/airflow/blob/master/airflow/models/pool.py#L49\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nfrom typing import Dict, Iterable, Optional, Tuple\n\nfrom sqlalchemy import Column, Integer, String, Text, func\nfrom sqlalchemy.orm.session import Session\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.models.base import Base\nfrom airflow.ti_deps.dependencies_states import EXECUTION_STATES\nfrom airflow.typing_compat import TypedDict\nfrom airflow.utils.session import provide_session\nfrom airflow.utils.sqlalchemy import nowait, with_row_locks\nfrom airflow.utils.state import State\n\n\nclass PoolStats(TypedDict):\n    \"\"\"Dictionary containing Pool Stats\"\"\"\n\n    total: int\n    running: int\n    queued: int\n    open: int\n\n\nclass Pool(Base):\n    \"\"\"the class to get Pool info.\"\"\"\n\n    __tablename__ = \"slot_pool\"\n\n    id = Column(Integer, primary_key=True)\n    pool = Column(String(256), unique=True)\n    # -1 for infinite\n    slots = Column(Integer, default=0)\n    description = Column(Text)\n\n    DEFAULT_POOL_NAME = 'default_pool'\n\n    def __repr__(self):\n        return str(self.pool)  # pylint: disable=E0012\n\n    @staticmethod\n    @provide_session\n    def get_pool(pool_name, session: Session = None):\n        \"\"\"\n        Get the Pool with specific pool name from the Pools.\n\n        :param pool_name: The pool name of the Pool to get.\n        :param session: SQLAlchemy ORM Session\n        :return: the pool object\n        \"\"\"\n        return session.query(Pool).filter(Pool.pool == pool_name).first()\n\n    @staticmethod\n    @provide_session\n    def get_default_pool(session: Session = None):\n        \"\"\"\n        Get the Pool of the default_pool from the Pools.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the pool object\n        \"\"\"\n        return Pool.get_pool(Pool.DEFAULT_POOL_NAME, session=session)\n\n    @staticmethod\n    @provide_session\n    def slots_stats(\n        *,\n        lock_rows: bool = False,\n        session: Session = None,\n    ) -> Dict[str, PoolStats]:\n        \"\"\"\n        Get Pool stats (Number of Running, Queued, Open & Total tasks)\n\n        If ``lock_rows`` is True, and the database engine in use supports the ``NOWAIT`` syntax, then a\n        non-blocking lock will be attempted -- if the lock is not available then SQLAlchemy will throw an\n        OperationalError.\n\n        :param lock_rows: Should we attempt to obtain a row-level lock on all the Pool rows returns\n        :param session: SQLAlchemy ORM Session\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        pools: Dict[str, PoolStats] = {}\n\n        query = session.query(Pool.pool, Pool.slots)\n\n        if lock_rows:\n            query = with_row_locks(query, session=session, **nowait(session))\n\n        pool_rows: Iterable[Tuple[str, int]] = query.all()\n        for (pool_name, total_slots) in pool_rows:\n            pools[pool_name] = PoolStats(total=total_slots, running=0, queued=0, open=0)\n\n        state_count_by_pool = (\n            session.query(TaskInstance.pool, TaskInstance.state, func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.state.in_(list(EXECUTION_STATES)))\n            .group_by(TaskInstance.pool, TaskInstance.state)\n        ).all()\n\n        # calculate queued and running metrics\n        count: int\n        for (pool_name, state, count) in state_count_by_pool:\n            stats_dict: Optional[PoolStats] = pools.get(pool_name)\n            if not stats_dict:\n                continue\n            # TypedDict key must be a string literal, so we use if-statements to set value\n            if state == \"running\":\n                stats_dict[\"running\"] = count\n            elif state == \"queued\":\n                stats_dict[\"queued\"] = count\n            else:\n                raise AirflowException(f\"Unexpected state. Expected values: {EXECUTION_STATES}.\")\n\n        # calculate open metric\n        for pool_name, stats_dict in pools.items():\n            if stats_dict[\"total\"] == -1:\n                # -1 means infinite\n                stats_dict[\"open\"] = -1\n            else:\n                stats_dict[\"open\"] = stats_dict[\"total\"] - stats_dict[\"running\"] - stats_dict[\"queued\"]\n\n        return pools\n\n    def to_json(self):\n        \"\"\"\n        Get the Pool in a json structure\n\n        :return: the pool object in json format\n        \"\"\"\n        return {\n            'id': self.id,\n            'pool': self.pool,\n            'slots': self.slots,\n            'description': self.description,\n        }\n\n    @provide_session\n    def occupied_slots(self, session: Session):\n        \"\"\"\n        Get the number of slots used by running/queued tasks at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the used number of slots\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        return int(\n            session.query(func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.pool == self.pool)\n            .filter(TaskInstance.state.in_(list(EXECUTION_STATES)))\n            .scalar()\n            or 0\n        )\n\n    @provide_session\n    def running_slots(self, session: Session):\n        \"\"\"\n        Get the number of slots used by running tasks at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the used number of slots\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        return int(\n            session.query(func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.pool == self.pool)\n            .filter(TaskInstance.state == State.RUNNING)\n            .scalar()\n            or 0\n        )\n\n    @provide_session\n    def queued_slots(self, session: Session):\n        \"\"\"\n        Get the number of slots used by queued tasks at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the used number of slots\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        return int(\n            session.query(func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.pool == self.pool)\n            .filter(TaskInstance.state == State.QUEUED)\n            .scalar()\n            or 0\n        )\n\n    @provide_session\n    def open_slots(self, session: Session) -> float:\n        \"\"\"\n        Get the number of slots open at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the number of slots\n        \"\"\"\n        if self.slots == -1:\n            return float('inf')\n        else:\n            return self.slots - self.occupied_slots(session)\n", "path": "airflow/models/pool.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nfrom typing import Dict, Iterable, Optional, Tuple\n\nfrom sqlalchemy import Column, Integer, String, Text, func\nfrom sqlalchemy.orm.session import Session\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.models.base import Base\nfrom airflow.ti_deps.dependencies_states import EXECUTION_STATES\nfrom airflow.typing_compat import TypedDict\nfrom airflow.utils.session import provide_session\nfrom airflow.utils.sqlalchemy import nowait, with_row_locks\nfrom airflow.utils.state import State\n\n\nclass PoolStats(TypedDict):\n    \"\"\"Dictionary containing Pool Stats\"\"\"\n\n    total: int\n    running: int\n    queued: int\n    open: int\n\n\nclass Pool(Base):\n    \"\"\"the class to get Pool info.\"\"\"\n\n    __tablename__ = \"slot_pool\"\n\n    id = Column(Integer, primary_key=True)\n    pool = Column(String(256), unique=True)\n    # -1 for infinite\n    slots = Column(Integer, default=0)\n    description = Column(Text)\n\n    DEFAULT_POOL_NAME = 'default_pool'\n\n    def __repr__(self):\n        return str(self.pool)  # pylint: disable=E0012\n\n    @staticmethod\n    @provide_session\n    def get_pool(pool_name, session: Session = None):\n        \"\"\"\n        Get the Pool with specific pool name from the Pools.\n\n        :param pool_name: The pool name of the Pool to get.\n        :param session: SQLAlchemy ORM Session\n        :return: the pool object\n        \"\"\"\n        return session.query(Pool).filter(Pool.pool == pool_name).first()\n\n    @staticmethod\n    @provide_session\n    def get_default_pool(session: Session = None):\n        \"\"\"\n        Get the Pool of the default_pool from the Pools.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the pool object\n        \"\"\"\n        return Pool.get_pool(Pool.DEFAULT_POOL_NAME, session=session)\n\n    @staticmethod\n    @provide_session\n    def slots_stats(\n        *,\n        lock_rows: bool = False,\n        session: Session = None,\n    ) -> Dict[str, PoolStats]:\n        \"\"\"\n        Get Pool stats (Number of Running, Queued, Open & Total tasks)\n\n        If ``lock_rows`` is True, and the database engine in use supports the ``NOWAIT`` syntax, then a\n        non-blocking lock will be attempted -- if the lock is not available then SQLAlchemy will throw an\n        OperationalError.\n\n        :param lock_rows: Should we attempt to obtain a row-level lock on all the Pool rows returns\n        :param session: SQLAlchemy ORM Session\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        pools: Dict[str, PoolStats] = {}\n\n        query = session.query(Pool.pool, Pool.slots)\n\n        if lock_rows:\n            query = with_row_locks(query, session=session, **nowait(session))\n\n        pool_rows: Iterable[Tuple[str, int]] = query.all()\n        for (pool_name, total_slots) in pool_rows:\n            if total_slots == -1:\n                total_slots = float('inf')  # type: ignore\n            pools[pool_name] = PoolStats(total=total_slots, running=0, queued=0, open=0)\n\n        state_count_by_pool = (\n            session.query(TaskInstance.pool, TaskInstance.state, func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.state.in_(list(EXECUTION_STATES)))\n            .group_by(TaskInstance.pool, TaskInstance.state)\n        ).all()\n\n        # calculate queued and running metrics\n        for (pool_name, state, count) in state_count_by_pool:\n            # Some databases return decimal.Decimal here.\n            count = int(count)\n\n            stats_dict: Optional[PoolStats] = pools.get(pool_name)\n            if not stats_dict:\n                continue\n            # TypedDict key must be a string literal, so we use if-statements to set value\n            if state == \"running\":\n                stats_dict[\"running\"] = count\n            elif state == \"queued\":\n                stats_dict[\"queued\"] = count\n            else:\n                raise AirflowException(f\"Unexpected state. Expected values: {EXECUTION_STATES}.\")\n\n        # calculate open metric\n        for pool_name, stats_dict in pools.items():\n            if stats_dict[\"total\"] == -1:\n                # -1 means infinite\n                stats_dict[\"open\"] = -1\n            else:\n                stats_dict[\"open\"] = stats_dict[\"total\"] - stats_dict[\"running\"] - stats_dict[\"queued\"]\n\n        return pools\n\n    def to_json(self):\n        \"\"\"\n        Get the Pool in a json structure\n\n        :return: the pool object in json format\n        \"\"\"\n        return {\n            'id': self.id,\n            'pool': self.pool,\n            'slots': self.slots,\n            'description': self.description,\n        }\n\n    @provide_session\n    def occupied_slots(self, session: Session):\n        \"\"\"\n        Get the number of slots used by running/queued tasks at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the used number of slots\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        return int(\n            session.query(func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.pool == self.pool)\n            .filter(TaskInstance.state.in_(list(EXECUTION_STATES)))\n            .scalar()\n            or 0\n        )\n\n    @provide_session\n    def running_slots(self, session: Session):\n        \"\"\"\n        Get the number of slots used by running tasks at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the used number of slots\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        return int(\n            session.query(func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.pool == self.pool)\n            .filter(TaskInstance.state == State.RUNNING)\n            .scalar()\n            or 0\n        )\n\n    @provide_session\n    def queued_slots(self, session: Session):\n        \"\"\"\n        Get the number of slots used by queued tasks at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the used number of slots\n        \"\"\"\n        from airflow.models.taskinstance import TaskInstance  # Avoid circular import\n\n        return int(\n            session.query(func.sum(TaskInstance.pool_slots))\n            .filter(TaskInstance.pool == self.pool)\n            .filter(TaskInstance.state == State.QUEUED)\n            .scalar()\n            or 0\n        )\n\n    @provide_session\n    def open_slots(self, session: Session) -> float:\n        \"\"\"\n        Get the number of slots open at the moment.\n\n        :param session: SQLAlchemy ORM Session\n        :return: the number of slots\n        \"\"\"\n        if self.slots == -1:\n            return float('inf')\n        else:\n            return self.slots - self.occupied_slots(session)\n", "path": "airflow/models/pool.py"}]}