| Server IP : 3.96.16.70 / Your IP : 216.73.216.15 Web Server : Apache System : Linux ip-172-31-26-103.ca-central-1.compute.internal 6.1.163-186.299.amzn2023.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Feb 24 16:35:42 UTC 2026 x86_64 User : ec2-user ( 1000) PHP Version : 8.4.18 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : OFF Directory : /lib/python3.9/site-packages/awscli/autocomplete/ |
Upload File : |
import logging
import os
import sqlite3
from awscli import __version__ as cli_version
LOG = logging.getLogger(__name__)
# We may eventually include a pre-generated version of this index as part
# of our shipped distributable, but for now we'll add this to our cache
# dir.
INDEX_DIR = os.path.expanduser(os.path.join('~', '.aws', 'cli', 'cache'))
INDEX_FILE = os.path.join(INDEX_DIR, '%s.index' % cli_version)
BUILTIN_INDEX_FILE = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
'data',
'ac.index',
)
# This is similar to DBConnection in awscli.customization.history.
# I'd like to reuse code, but we also have the contraint that we don't
# want to import anything outside of awscli.autocomplete to ensure
# our startup time is as minimal as possible.
class DatabaseConnection:
_JOURNAL_MODE_OFF = 'PRAGMA journal_mode=OFF'
def __init__(self, db_filename=None):
self._db_conn = None
self._db_filename = db_filename
if self._db_filename is None:
self._db_filename = self._get_index_filename()
@property
def _connection(self):
if self._db_conn is None:
kwargs = {'check_same_thread': False, 'isolation_level': None}
if self._db_filename.startswith('file::memory:'):
# This statement was added because old versions of sqlite
# don't support 'uri' but we use it for tests and because
# in the tests we use only in-memory database we're looking
# for 'file::memory:' prefix if we decide to use 'uri' with
# sqlite more widely we can ease restrictions to 'file:'
kwargs['uri'] = True
self._db_conn = sqlite3.connect(self._db_filename, **kwargs)
self._ensure_database_setup()
return self._db_conn
def close(self):
self._connection.close()
def execute(self, query, **kwargs):
return self._connection.execute(query, kwargs)
def _ensure_database_setup(self):
# We are setting journal mode to off because there is no guarantee
# the user has permissions to write temporary files to where the
# CLI is installed and in-practice, the index is only ever read from
# (except when we need to generate it).
self.execute(self._JOURNAL_MODE_OFF)
def _get_index_filename(self):
if os.path.isfile(INDEX_FILE):
return INDEX_FILE
return BUILTIN_INDEX_FILE