We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
 cannot import name 'clock_gettime' from 'time' (unknown location)
### Expected behaviour
Import time

### Actual behaviour

  File "C:\ProgramData\Anaconda3\lib\site-packages\datacube\drivers\postgres\_connections.py", line 20, in <module>
    from time import clock_gettime, CLOCK_REALTIME

ImportError: cannot import name 'clock_gettime' from 'time' (unknown location)

cannot import name 'clock_gettime' from 'time' (unknown location)


### Steps to reproduce the behaviour
  from ._connections import PostgresDb

### Environment information
Python 3.8.11
* Which ``datacube --version`` are you using?
Unknown, it's referenced in geocube on conda-forge
* What datacube deployment/enviornment are you running against?
n/a

Additional notes:
time.clock has been removed in Python 3.8 after being deprecated in Python 3.3. The recommendation is to change clock by either time.perf_counter or time.process_time.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `datacube/drivers/postgres/_connections.py`
Content:
```
1 # This file is part of the Open Data Cube, see https://opendatacube.org for more information
2 #
3 # Copyright (c) 2015-2020 ODC Contributors
4 # SPDX-License-Identifier: Apache-2.0
5 
6 # We often have one-arg-per column, so these checks aren't so useful.
7 # pylint: disable=too-many-arguments,too-many-public-methods
8 
9 # SQLAlchemy queries require "column == None", not "column is None" due to operator overloading:
10 # pylint: disable=singleton-comparison
11 
12 """
13 Postgres connection and setup
14 """
15 import json
16 import logging
17 import os
18 import re
19 from contextlib import contextmanager
20 from time import clock_gettime, CLOCK_REALTIME
21 from typing import Callable, Optional, Union
22 
23 from sqlalchemy import event, create_engine, text
24 from sqlalchemy.engine import Engine
25 from sqlalchemy.engine.url import URL as EngineUrl
26 
27 import datacube
28 from datacube.index.exceptions import IndexSetupError
29 from datacube.utils import jsonify_document
30 
31 from . import _api
32 from . import _core
33 
34 _LIB_ID = 'agdc-' + str(datacube.__version__)
35 
36 _LOG = logging.getLogger(__name__)
37 
38 try:
39     import pwd
40 
41     DEFAULT_DB_USER = pwd.getpwuid(os.geteuid()).pw_name  # type: Optional[str]
42 except (ImportError, KeyError):
43     # No default on Windows and some other systems
44     DEFAULT_DB_USER = None
45 DEFAULT_DB_PORT = 5432
46 DEFAULT_IAM_AUTH = False
47 DEFAULT_IAM_TIMEOUT = 600
48 
49 
50 class PostgresDb(object):
51     """
52     A thin database access api.
53 
54     It exists so that higher level modules are not tied to SQLAlchemy, connections or specifics of database-access.
55 
56     (and can be unit tested without any actual databases)
57 
58     Thread safe: the only shared state is the (thread-safe) sqlalchemy connection pool.
59 
60     But not multiprocess safe once the first connections are made! A connection must not be shared between multiple
61     processes. You can call close() before forking if you know no other threads currently hold connections,
62     or else use a separate instance of this class in each process.
63     """
64 
65     def __init__(self, engine):
66         # We don't recommend using this constructor directly as it may change.
67         # Use static methods PostgresDb.create() or PostgresDb.from_config()
68         self._engine = engine
69 
70     @classmethod
71     def from_config(cls, config, application_name=None, validate_connection=True):
72         app_name = cls._expand_app_name(application_name)
73 
74         return PostgresDb.create(
75             config['db_hostname'],
76             config['db_database'],
77             config.get('db_username', DEFAULT_DB_USER),
78             config.get('db_password', None),
79             config.get('db_port', DEFAULT_DB_PORT),
80             application_name=app_name,
81             validate=validate_connection,
82             iam_rds_auth=bool(config.get("db_iam_authentication", DEFAULT_IAM_AUTH)),
83             iam_rds_timeout=int(config.get("db_iam_timeout", DEFAULT_IAM_TIMEOUT)),
84             pool_timeout=int(config.get('db_connection_timeout', 60)),
85             # pass config?
86         )
87 
88     @classmethod
89     def create(cls, hostname, database, username=None, password=None, port=None,
90                application_name=None, validate=True,
91                iam_rds_auth=False, iam_rds_timeout=600,
92                # pass config?
93                pool_timeout=60):
94         mk_url = getattr(EngineUrl, 'create', EngineUrl)
95         engine = cls._create_engine(
96             mk_url(
97                 'postgresql',
98                 host=hostname, database=database, port=port,
99                 username=username, password=password,
100             ),
101             application_name=application_name,
102             iam_rds_auth=iam_rds_auth,
103             iam_rds_timeout=iam_rds_timeout,
104             pool_timeout=pool_timeout)
105         if validate:
106             if not _core.database_exists(engine):
107                 raise IndexSetupError('\n\nNo DB schema exists. Have you run init?\n\t{init_command}'.format(
108                     init_command='datacube system init'
109                 ))
110 
111             if not _core.schema_is_latest(engine):
112                 raise IndexSetupError(
113                     '\n\nDB schema is out of date. '
114                     'An administrator must run init:\n\t{init_command}'.format(
115                         init_command='datacube -v system init'
116                     ))
117         return PostgresDb(engine)
118 
119     @staticmethod
120     def _create_engine(url, application_name=None, iam_rds_auth=False, iam_rds_timeout=600, pool_timeout=60):
121         engine = create_engine(
122             url,
123             echo=False,
124             echo_pool=False,
125 
126             # 'AUTOCOMMIT' here means READ-COMMITTED isolation level with autocommit on.
127             # When a transaction is needed we will do an explicit begin/commit.
128             isolation_level='AUTOCOMMIT',
129             json_serializer=_to_json,
130             # If a connection is idle for this many seconds, SQLAlchemy will renew it rather
131             # than assuming it's still open. Allows servers to close idle connections without clients
132             # getting errors.
133             pool_recycle=pool_timeout,
134             connect_args={'application_name': application_name}
135         )
136 
137         if iam_rds_auth:
138             from datacube.utils.aws import obtain_new_iam_auth_token
139             handle_dynamic_token_authentication(engine, obtain_new_iam_auth_token, timeout=iam_rds_timeout, url=url)
140 
141         return engine
142 
143     @property
144     def url(self) -> EngineUrl:
145         return self._engine.url
146 
147     @staticmethod
148     def get_db_username(config):
149         try:
150             return config['db_username']
151         except KeyError:
152             return DEFAULT_DB_USER
153 
154     def close(self):
155         """
156         Close any idle connections in the pool.
157 
158         This is good practice if you are keeping this object in scope
159         but wont be using it for a while.
160 
161         Connections should not be shared between processes, so this should be called
162         before forking if the same instance will be used.
163 
164         (connections are normally closed automatically when this object is
165          garbage collected)
166         """
167         self._engine.dispose()
168 
169     @classmethod
170     def _expand_app_name(cls, application_name):
171         """
172         >>> PostgresDb._expand_app_name(None) #doctest: +ELLIPSIS
173         'agdc-...'
174         >>> PostgresDb._expand_app_name('') #doctest: +ELLIPSIS
175         'agdc-...'
176         >>> PostgresDb._expand_app_name('cli') #doctest: +ELLIPSIS
177         'cli agdc-...'
178         >>> PostgresDb._expand_app_name('a b.c/d')
179         'a-b-c-d agdc-...'
180         >>> PostgresDb._expand_app_name(5)
181         Traceback (most recent call last):
182         ...
183         TypeError: Application name must be a string
184         """
185         full_name = _LIB_ID
186         if application_name:
187             if not isinstance(application_name, str):
188                 raise TypeError('Application name must be a string')
189 
190             full_name = re.sub('[^0-9a-zA-Z]+', '-', application_name) + ' ' + full_name
191 
192         if len(full_name) > 64:
193             _LOG.warning('Application name is too long: Truncating to %s chars', (64 - len(_LIB_ID) - 1))
194         return full_name[-64:]
195 
196     def init(self, with_permissions=True):
197         """
198         Init a new database (if not already set up).
199 
200         :return: If it was newly created.
201         """
202         is_new = _core.ensure_db(self._engine, with_permissions=with_permissions)
203         if not is_new:
204             _core.update_schema(self._engine)
205 
206         return is_new
207 
208     @contextmanager
209     def connect(self):
210         """
211         Borrow a connection from the pool.
212 
213         The name connect() is misleading: it will not create a new connection if one is already available in the pool.
214 
215         Callers should minimise the amount of time they hold onto their connections. If they're doing anything between
216         calls to the DB (such as opening files, or waiting on user input), it's better to return the connection
217         to the pool beforehand.
218 
219         The connection can raise errors if not following this advice ("server closed the connection unexpectedly"),
220         as some servers will aggressively close idle connections (eg. DEA's NCI servers). It also prevents the
221         connection from being reused while borrowed.
222         """
223         with self._engine.connect() as connection:
224             yield _api.PostgresDbAPI(connection)
225             connection.close()
226 
227     @contextmanager
228     def begin(self):
229         """
230         Start a transaction.
231 
232         Returns an instance that will maintain a single connection in a transaction.
233 
234         Call commit() or rollback() to complete the transaction or use a context manager:
235 
236             with db.begin() as trans:
237                 trans.insert_dataset(...)
238 
239         (Don't share an instance between threads)
240 
241         :rtype: PostgresDBAPI
242         """
243         with self._engine.connect() as connection:
244             connection.execute(text('BEGIN'))
245             try:
246                 yield _api.PostgresDbAPI(connection)
247                 connection.execute(text('COMMIT'))
248             except Exception:  # pylint: disable=broad-except
249                 connection.execute(text('ROLLBACK'))
250                 raise
251             finally:
252                 connection.close()
253 
254     def give_me_a_connection(self):
255         return self._engine.connect()
256 
257     @classmethod
258     def get_dataset_fields(cls, metadata_type_definition):
259         return _api.get_dataset_fields(metadata_type_definition)
260 
261     def __repr__(self):
262         return "PostgresDb<engine={!r}>".format(self._engine)
263 
264 
265 def handle_dynamic_token_authentication(engine: Engine,
266                                         new_token: Callable[..., str],
267                                         timeout: Union[float, int] = 600,
268                                         **kwargs) -> None:
269     last_token = [None]
270     last_token_time = [0.0]
271 
272     @event.listens_for(engine, "do_connect")
273     def override_new_connection(dialect, conn_rec, cargs, cparams):
274         # Handle IAM authentication
275         now = clock_gettime(CLOCK_REALTIME)
276         if now - last_token_time[0] > timeout:
277             last_token[0] = new_token(**kwargs)
278             last_token_time[0] = now
279         cparams["password"] = last_token[0]
280 
281 
282 def _to_json(o):
283     # Postgres <=9.5 doesn't support NaN and Infinity
284     fixedup = jsonify_document(o)
285     return json.dumps(fixedup, default=_json_fallback)
286 
287 
288 def _json_fallback(obj):
289     """Fallback json serialiser."""
290     raise TypeError("Type not serializable: {}".format(type(obj)))
291
```
--- 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/datacube/drivers/postgres/_connections.py b/datacube/drivers/postgres/_connections.py
--- a/datacube/drivers/postgres/_connections.py
+++ b/datacube/drivers/postgres/_connections.py
@@ -17,7 +17,6 @@
 import os
 import re
 from contextlib import contextmanager
-from time import clock_gettime, CLOCK_REALTIME
 from typing import Callable, Optional, Union
 
 from sqlalchemy import event, create_engine, text
@@ -272,6 +271,10 @@
     @event.listens_for(engine, "do_connect")
     def override_new_connection(dialect, conn_rec, cargs, cparams):
         # Handle IAM authentication
+        # Importing here because the function `clock_gettime` is not available on Windows
+        # which shouldn't be a problem, because boto3 auth is mostly used on AWS.
+        from time import clock_gettime, CLOCK_REALTIME
+
         now = clock_gettime(CLOCK_REALTIME)
         if now - last_token_time[0] > timeout:
             last_token[0] = new_token(**kwargs)

{"golden_diff": "diff --git a/datacube/drivers/postgres/_connections.py b/datacube/drivers/postgres/_connections.py\n--- a/datacube/drivers/postgres/_connections.py\n+++ b/datacube/drivers/postgres/_connections.py\n@@ -17,7 +17,6 @@\n import os\n import re\n from contextlib import contextmanager\n-from time import clock_gettime, CLOCK_REALTIME\n from typing import Callable, Optional, Union\n \n from sqlalchemy import event, create_engine, text\n@@ -272,6 +271,10 @@\n     @event.listens_for(engine, \"do_connect\")\n     def override_new_connection(dialect, conn_rec, cargs, cparams):\n         # Handle IAM authentication\n+        # Importing here because the function `clock_gettime` is not available on Windows\n+        # which shouldn't be a problem, because boto3 auth is mostly used on AWS.\n+        from time import clock_gettime, CLOCK_REALTIME\n+\n         now = clock_gettime(CLOCK_REALTIME)\n         if now - last_token_time[0] > timeout:\n             last_token[0] = new_token(**kwargs)\n", "issue": " cannot import name 'clock_gettime' from 'time' (unknown location)\n### Expected behaviour\r\nImport time\r\n\r\n### Actual behaviour\r\n\r\n  File \"C:\\ProgramData\\Anaconda3\\lib\\site-packages\\datacube\\drivers\\postgres\\_connections.py\", line 20, in <module>\r\n    from time import clock_gettime, CLOCK_REALTIME\r\n\r\nImportError: cannot import name 'clock_gettime' from 'time' (unknown location)\r\n\r\ncannot import name 'clock_gettime' from 'time' (unknown location)\r\n\r\n\r\n### Steps to reproduce the behaviour\r\n  from ._connections import PostgresDb\r\n\r\n### Environment information\r\nPython 3.8.11\r\n* Which ``datacube --version`` are you using?\r\nUnknown, it's referenced in geocube on conda-forge\r\n* What datacube deployment/enviornment are you running against?\r\nn/a\r\n\r\nAdditional notes:\r\ntime.clock has been removed in Python 3.8 after being deprecated in Python 3.3. The recommendation is to change clock by either time.perf_counter or time.process_time.\r\n\n", "before_files": [{"content": "# This file is part of the Open Data Cube, see https://opendatacube.org for more information\n#\n# Copyright (c) 2015-2020 ODC Contributors\n# SPDX-License-Identifier: Apache-2.0\n\n# We often have one-arg-per column, so these checks aren't so useful.\n# pylint: disable=too-many-arguments,too-many-public-methods\n\n# SQLAlchemy queries require \"column == None\", not \"column is None\" due to operator overloading:\n# pylint: disable=singleton-comparison\n\n\"\"\"\nPostgres connection and setup\n\"\"\"\nimport json\nimport logging\nimport os\nimport re\nfrom contextlib import contextmanager\nfrom time import clock_gettime, CLOCK_REALTIME\nfrom typing import Callable, Optional, Union\n\nfrom sqlalchemy import event, create_engine, text\nfrom sqlalchemy.engine import Engine\nfrom sqlalchemy.engine.url import URL as EngineUrl\n\nimport datacube\nfrom datacube.index.exceptions import IndexSetupError\nfrom datacube.utils import jsonify_document\n\nfrom . import _api\nfrom . import _core\n\n_LIB_ID = 'agdc-' + str(datacube.__version__)\n\n_LOG = logging.getLogger(__name__)\n\ntry:\n    import pwd\n\n    DEFAULT_DB_USER = pwd.getpwuid(os.geteuid()).pw_name  # type: Optional[str]\nexcept (ImportError, KeyError):\n    # No default on Windows and some other systems\n    DEFAULT_DB_USER = None\nDEFAULT_DB_PORT = 5432\nDEFAULT_IAM_AUTH = False\nDEFAULT_IAM_TIMEOUT = 600\n\n\nclass PostgresDb(object):\n    \"\"\"\n    A thin database access api.\n\n    It exists so that higher level modules are not tied to SQLAlchemy, connections or specifics of database-access.\n\n    (and can be unit tested without any actual databases)\n\n    Thread safe: the only shared state is the (thread-safe) sqlalchemy connection pool.\n\n    But not multiprocess safe once the first connections are made! A connection must not be shared between multiple\n    processes. You can call close() before forking if you know no other threads currently hold connections,\n    or else use a separate instance of this class in each process.\n    \"\"\"\n\n    def __init__(self, engine):\n        # We don't recommend using this constructor directly as it may change.\n        # Use static methods PostgresDb.create() or PostgresDb.from_config()\n        self._engine = engine\n\n    @classmethod\n    def from_config(cls, config, application_name=None, validate_connection=True):\n        app_name = cls._expand_app_name(application_name)\n\n        return PostgresDb.create(\n            config['db_hostname'],\n            config['db_database'],\n            config.get('db_username', DEFAULT_DB_USER),\n            config.get('db_password', None),\n            config.get('db_port', DEFAULT_DB_PORT),\n            application_name=app_name,\n            validate=validate_connection,\n            iam_rds_auth=bool(config.get(\"db_iam_authentication\", DEFAULT_IAM_AUTH)),\n            iam_rds_timeout=int(config.get(\"db_iam_timeout\", DEFAULT_IAM_TIMEOUT)),\n            pool_timeout=int(config.get('db_connection_timeout', 60)),\n            # pass config?\n        )\n\n    @classmethod\n    def create(cls, hostname, database, username=None, password=None, port=None,\n               application_name=None, validate=True,\n               iam_rds_auth=False, iam_rds_timeout=600,\n               # pass config?\n               pool_timeout=60):\n        mk_url = getattr(EngineUrl, 'create', EngineUrl)\n        engine = cls._create_engine(\n            mk_url(\n                'postgresql',\n                host=hostname, database=database, port=port,\n                username=username, password=password,\n            ),\n            application_name=application_name,\n            iam_rds_auth=iam_rds_auth,\n            iam_rds_timeout=iam_rds_timeout,\n            pool_timeout=pool_timeout)\n        if validate:\n            if not _core.database_exists(engine):\n                raise IndexSetupError('\\n\\nNo DB schema exists. Have you run init?\\n\\t{init_command}'.format(\n                    init_command='datacube system init'\n                ))\n\n            if not _core.schema_is_latest(engine):\n                raise IndexSetupError(\n                    '\\n\\nDB schema is out of date. '\n                    'An administrator must run init:\\n\\t{init_command}'.format(\n                        init_command='datacube -v system init'\n                    ))\n        return PostgresDb(engine)\n\n    @staticmethod\n    def _create_engine(url, application_name=None, iam_rds_auth=False, iam_rds_timeout=600, pool_timeout=60):\n        engine = create_engine(\n            url,\n            echo=False,\n            echo_pool=False,\n\n            # 'AUTOCOMMIT' here means READ-COMMITTED isolation level with autocommit on.\n            # When a transaction is needed we will do an explicit begin/commit.\n            isolation_level='AUTOCOMMIT',\n            json_serializer=_to_json,\n            # If a connection is idle for this many seconds, SQLAlchemy will renew it rather\n            # than assuming it's still open. Allows servers to close idle connections without clients\n            # getting errors.\n            pool_recycle=pool_timeout,\n            connect_args={'application_name': application_name}\n        )\n\n        if iam_rds_auth:\n            from datacube.utils.aws import obtain_new_iam_auth_token\n            handle_dynamic_token_authentication(engine, obtain_new_iam_auth_token, timeout=iam_rds_timeout, url=url)\n\n        return engine\n\n    @property\n    def url(self) -> EngineUrl:\n        return self._engine.url\n\n    @staticmethod\n    def get_db_username(config):\n        try:\n            return config['db_username']\n        except KeyError:\n            return DEFAULT_DB_USER\n\n    def close(self):\n        \"\"\"\n        Close any idle connections in the pool.\n\n        This is good practice if you are keeping this object in scope\n        but wont be using it for a while.\n\n        Connections should not be shared between processes, so this should be called\n        before forking if the same instance will be used.\n\n        (connections are normally closed automatically when this object is\n         garbage collected)\n        \"\"\"\n        self._engine.dispose()\n\n    @classmethod\n    def _expand_app_name(cls, application_name):\n        \"\"\"\n        >>> PostgresDb._expand_app_name(None) #doctest: +ELLIPSIS\n        'agdc-...'\n        >>> PostgresDb._expand_app_name('') #doctest: +ELLIPSIS\n        'agdc-...'\n        >>> PostgresDb._expand_app_name('cli') #doctest: +ELLIPSIS\n        'cli agdc-...'\n        >>> PostgresDb._expand_app_name('a b.c/d')\n        'a-b-c-d agdc-...'\n        >>> PostgresDb._expand_app_name(5)\n        Traceback (most recent call last):\n        ...\n        TypeError: Application name must be a string\n        \"\"\"\n        full_name = _LIB_ID\n        if application_name:\n            if not isinstance(application_name, str):\n                raise TypeError('Application name must be a string')\n\n            full_name = re.sub('[^0-9a-zA-Z]+', '-', application_name) + ' ' + full_name\n\n        if len(full_name) > 64:\n            _LOG.warning('Application name is too long: Truncating to %s chars', (64 - len(_LIB_ID) - 1))\n        return full_name[-64:]\n\n    def init(self, with_permissions=True):\n        \"\"\"\n        Init a new database (if not already set up).\n\n        :return: If it was newly created.\n        \"\"\"\n        is_new = _core.ensure_db(self._engine, with_permissions=with_permissions)\n        if not is_new:\n            _core.update_schema(self._engine)\n\n        return is_new\n\n    @contextmanager\n    def connect(self):\n        \"\"\"\n        Borrow a connection from the pool.\n\n        The name connect() is misleading: it will not create a new connection if one is already available in the pool.\n\n        Callers should minimise the amount of time they hold onto their connections. If they're doing anything between\n        calls to the DB (such as opening files, or waiting on user input), it's better to return the connection\n        to the pool beforehand.\n\n        The connection can raise errors if not following this advice (\"server closed the connection unexpectedly\"),\n        as some servers will aggressively close idle connections (eg. DEA's NCI servers). It also prevents the\n        connection from being reused while borrowed.\n        \"\"\"\n        with self._engine.connect() as connection:\n            yield _api.PostgresDbAPI(connection)\n            connection.close()\n\n    @contextmanager\n    def begin(self):\n        \"\"\"\n        Start a transaction.\n\n        Returns an instance that will maintain a single connection in a transaction.\n\n        Call commit() or rollback() to complete the transaction or use a context manager:\n\n            with db.begin() as trans:\n                trans.insert_dataset(...)\n\n        (Don't share an instance between threads)\n\n        :rtype: PostgresDBAPI\n        \"\"\"\n        with self._engine.connect() as connection:\n            connection.execute(text('BEGIN'))\n            try:\n                yield _api.PostgresDbAPI(connection)\n                connection.execute(text('COMMIT'))\n            except Exception:  # pylint: disable=broad-except\n                connection.execute(text('ROLLBACK'))\n                raise\n            finally:\n                connection.close()\n\n    def give_me_a_connection(self):\n        return self._engine.connect()\n\n    @classmethod\n    def get_dataset_fields(cls, metadata_type_definition):\n        return _api.get_dataset_fields(metadata_type_definition)\n\n    def __repr__(self):\n        return \"PostgresDb<engine={!r}>\".format(self._engine)\n\n\ndef handle_dynamic_token_authentication(engine: Engine,\n                                        new_token: Callable[..., str],\n                                        timeout: Union[float, int] = 600,\n                                        **kwargs) -> None:\n    last_token = [None]\n    last_token_time = [0.0]\n\n    @event.listens_for(engine, \"do_connect\")\n    def override_new_connection(dialect, conn_rec, cargs, cparams):\n        # Handle IAM authentication\n        now = clock_gettime(CLOCK_REALTIME)\n        if now - last_token_time[0] > timeout:\n            last_token[0] = new_token(**kwargs)\n            last_token_time[0] = now\n        cparams[\"password\"] = last_token[0]\n\n\ndef _to_json(o):\n    # Postgres <=9.5 doesn't support NaN and Infinity\n    fixedup = jsonify_document(o)\n    return json.dumps(fixedup, default=_json_fallback)\n\n\ndef _json_fallback(obj):\n    \"\"\"Fallback json serialiser.\"\"\"\n    raise TypeError(\"Type not serializable: {}\".format(type(obj)))\n", "path": "datacube/drivers/postgres/_connections.py"}], "after_files": [{"content": "# This file is part of the Open Data Cube, see https://opendatacube.org for more information\n#\n# Copyright (c) 2015-2020 ODC Contributors\n# SPDX-License-Identifier: Apache-2.0\n\n# We often have one-arg-per column, so these checks aren't so useful.\n# pylint: disable=too-many-arguments,too-many-public-methods\n\n# SQLAlchemy queries require \"column == None\", not \"column is None\" due to operator overloading:\n# pylint: disable=singleton-comparison\n\n\"\"\"\nPostgres connection and setup\n\"\"\"\nimport json\nimport logging\nimport os\nimport re\nfrom contextlib import contextmanager\nfrom typing import Callable, Optional, Union\n\nfrom sqlalchemy import event, create_engine, text\nfrom sqlalchemy.engine import Engine\nfrom sqlalchemy.engine.url import URL as EngineUrl\n\nimport datacube\nfrom datacube.index.exceptions import IndexSetupError\nfrom datacube.utils import jsonify_document\n\nfrom . import _api\nfrom . import _core\n\n_LIB_ID = 'agdc-' + str(datacube.__version__)\n\n_LOG = logging.getLogger(__name__)\n\ntry:\n    import pwd\n\n    DEFAULT_DB_USER = pwd.getpwuid(os.geteuid()).pw_name  # type: Optional[str]\nexcept (ImportError, KeyError):\n    # No default on Windows and some other systems\n    DEFAULT_DB_USER = None\nDEFAULT_DB_PORT = 5432\nDEFAULT_IAM_AUTH = False\nDEFAULT_IAM_TIMEOUT = 600\n\n\nclass PostgresDb(object):\n    \"\"\"\n    A thin database access api.\n\n    It exists so that higher level modules are not tied to SQLAlchemy, connections or specifics of database-access.\n\n    (and can be unit tested without any actual databases)\n\n    Thread safe: the only shared state is the (thread-safe) sqlalchemy connection pool.\n\n    But not multiprocess safe once the first connections are made! A connection must not be shared between multiple\n    processes. You can call close() before forking if you know no other threads currently hold connections,\n    or else use a separate instance of this class in each process.\n    \"\"\"\n\n    def __init__(self, engine):\n        # We don't recommend using this constructor directly as it may change.\n        # Use static methods PostgresDb.create() or PostgresDb.from_config()\n        self._engine = engine\n\n    @classmethod\n    def from_config(cls, config, application_name=None, validate_connection=True):\n        app_name = cls._expand_app_name(application_name)\n\n        return PostgresDb.create(\n            config['db_hostname'],\n            config['db_database'],\n            config.get('db_username', DEFAULT_DB_USER),\n            config.get('db_password', None),\n            config.get('db_port', DEFAULT_DB_PORT),\n            application_name=app_name,\n            validate=validate_connection,\n            iam_rds_auth=bool(config.get(\"db_iam_authentication\", DEFAULT_IAM_AUTH)),\n            iam_rds_timeout=int(config.get(\"db_iam_timeout\", DEFAULT_IAM_TIMEOUT)),\n            pool_timeout=int(config.get('db_connection_timeout', 60)),\n            # pass config?\n        )\n\n    @classmethod\n    def create(cls, hostname, database, username=None, password=None, port=None,\n               application_name=None, validate=True,\n               iam_rds_auth=False, iam_rds_timeout=600,\n               # pass config?\n               pool_timeout=60):\n        mk_url = getattr(EngineUrl, 'create', EngineUrl)\n        engine = cls._create_engine(\n            mk_url(\n                'postgresql',\n                host=hostname, database=database, port=port,\n                username=username, password=password,\n            ),\n            application_name=application_name,\n            iam_rds_auth=iam_rds_auth,\n            iam_rds_timeout=iam_rds_timeout,\n            pool_timeout=pool_timeout)\n        if validate:\n            if not _core.database_exists(engine):\n                raise IndexSetupError('\\n\\nNo DB schema exists. Have you run init?\\n\\t{init_command}'.format(\n                    init_command='datacube system init'\n                ))\n\n            if not _core.schema_is_latest(engine):\n                raise IndexSetupError(\n                    '\\n\\nDB schema is out of date. '\n                    'An administrator must run init:\\n\\t{init_command}'.format(\n                        init_command='datacube -v system init'\n                    ))\n        return PostgresDb(engine)\n\n    @staticmethod\n    def _create_engine(url, application_name=None, iam_rds_auth=False, iam_rds_timeout=600, pool_timeout=60):\n        engine = create_engine(\n            url,\n            echo=False,\n            echo_pool=False,\n\n            # 'AUTOCOMMIT' here means READ-COMMITTED isolation level with autocommit on.\n            # When a transaction is needed we will do an explicit begin/commit.\n            isolation_level='AUTOCOMMIT',\n            json_serializer=_to_json,\n            # If a connection is idle for this many seconds, SQLAlchemy will renew it rather\n            # than assuming it's still open. Allows servers to close idle connections without clients\n            # getting errors.\n            pool_recycle=pool_timeout,\n            connect_args={'application_name': application_name}\n        )\n\n        if iam_rds_auth:\n            from datacube.utils.aws import obtain_new_iam_auth_token\n            handle_dynamic_token_authentication(engine, obtain_new_iam_auth_token, timeout=iam_rds_timeout, url=url)\n\n        return engine\n\n    @property\n    def url(self) -> EngineUrl:\n        return self._engine.url\n\n    @staticmethod\n    def get_db_username(config):\n        try:\n            return config['db_username']\n        except KeyError:\n            return DEFAULT_DB_USER\n\n    def close(self):\n        \"\"\"\n        Close any idle connections in the pool.\n\n        This is good practice if you are keeping this object in scope\n        but wont be using it for a while.\n\n        Connections should not be shared between processes, so this should be called\n        before forking if the same instance will be used.\n\n        (connections are normally closed automatically when this object is\n         garbage collected)\n        \"\"\"\n        self._engine.dispose()\n\n    @classmethod\n    def _expand_app_name(cls, application_name):\n        \"\"\"\n        >>> PostgresDb._expand_app_name(None) #doctest: +ELLIPSIS\n        'agdc-...'\n        >>> PostgresDb._expand_app_name('') #doctest: +ELLIPSIS\n        'agdc-...'\n        >>> PostgresDb._expand_app_name('cli') #doctest: +ELLIPSIS\n        'cli agdc-...'\n        >>> PostgresDb._expand_app_name('a b.c/d')\n        'a-b-c-d agdc-...'\n        >>> PostgresDb._expand_app_name(5)\n        Traceback (most recent call last):\n        ...\n        TypeError: Application name must be a string\n        \"\"\"\n        full_name = _LIB_ID\n        if application_name:\n            if not isinstance(application_name, str):\n                raise TypeError('Application name must be a string')\n\n            full_name = re.sub('[^0-9a-zA-Z]+', '-', application_name) + ' ' + full_name\n\n        if len(full_name) > 64:\n            _LOG.warning('Application name is too long: Truncating to %s chars', (64 - len(_LIB_ID) - 1))\n        return full_name[-64:]\n\n    def init(self, with_permissions=True):\n        \"\"\"\n        Init a new database (if not already set up).\n\n        :return: If it was newly created.\n        \"\"\"\n        is_new = _core.ensure_db(self._engine, with_permissions=with_permissions)\n        if not is_new:\n            _core.update_schema(self._engine)\n\n        return is_new\n\n    @contextmanager\n    def connect(self):\n        \"\"\"\n        Borrow a connection from the pool.\n\n        The name connect() is misleading: it will not create a new connection if one is already available in the pool.\n\n        Callers should minimise the amount of time they hold onto their connections. If they're doing anything between\n        calls to the DB (such as opening files, or waiting on user input), it's better to return the connection\n        to the pool beforehand.\n\n        The connection can raise errors if not following this advice (\"server closed the connection unexpectedly\"),\n        as some servers will aggressively close idle connections (eg. DEA's NCI servers). It also prevents the\n        connection from being reused while borrowed.\n        \"\"\"\n        with self._engine.connect() as connection:\n            yield _api.PostgresDbAPI(connection)\n            connection.close()\n\n    @contextmanager\n    def begin(self):\n        \"\"\"\n        Start a transaction.\n\n        Returns an instance that will maintain a single connection in a transaction.\n\n        Call commit() or rollback() to complete the transaction or use a context manager:\n\n            with db.begin() as trans:\n                trans.insert_dataset(...)\n\n        (Don't share an instance between threads)\n\n        :rtype: PostgresDBAPI\n        \"\"\"\n        with self._engine.connect() as connection:\n            connection.execute(text('BEGIN'))\n            try:\n                yield _api.PostgresDbAPI(connection)\n                connection.execute(text('COMMIT'))\n            except Exception:  # pylint: disable=broad-except\n                connection.execute(text('ROLLBACK'))\n                raise\n            finally:\n                connection.close()\n\n    def give_me_a_connection(self):\n        return self._engine.connect()\n\n    @classmethod\n    def get_dataset_fields(cls, metadata_type_definition):\n        return _api.get_dataset_fields(metadata_type_definition)\n\n    def __repr__(self):\n        return \"PostgresDb<engine={!r}>\".format(self._engine)\n\n\ndef handle_dynamic_token_authentication(engine: Engine,\n                                        new_token: Callable[..., str],\n                                        timeout: Union[float, int] = 600,\n                                        **kwargs) -> None:\n    last_token = [None]\n    last_token_time = [0.0]\n\n    @event.listens_for(engine, \"do_connect\")\n    def override_new_connection(dialect, conn_rec, cargs, cparams):\n        # Handle IAM authentication\n        # Importing here because the function `clock_gettime` is not available on Windows\n        # which shouldn't be a problem, because boto3 auth is mostly used on AWS.\n        from time import clock_gettime, CLOCK_REALTIME\n\n        now = clock_gettime(CLOCK_REALTIME)\n        if now - last_token_time[0] > timeout:\n            last_token[0] = new_token(**kwargs)\n            last_token_time[0] = now\n        cparams[\"password\"] = last_token[0]\n\n\ndef _to_json(o):\n    # Postgres <=9.5 doesn't support NaN and Infinity\n    fixedup = jsonify_document(o)\n    return json.dumps(fixedup, default=_json_fallback)\n\n\ndef _json_fallback(obj):\n    \"\"\"Fallback json serialiser.\"\"\"\n    raise TypeError(\"Type not serializable: {}\".format(type(obj)))\n", "path": "datacube/drivers/postgres/_connections.py"}]}