Skip to content

Instantly share code, notes, and snippets.

@shpaker
Created December 7, 2022 08:04
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save shpaker/ad563fbaecaf2b07817013449a1a0994 to your computer and use it in GitHub Desktop.
Save shpaker/ad563fbaecaf2b07817013449a1a0994 to your computer and use it in GitHub Desktop.
Скриптик для сбора базовой инфы о том, что лежит в монге
# В общем, на хосте где есть третий питон и доступ к этой монге надо поставить пакеты pymongo и click для скрипта:
#
# pip3 install click pymongo -U
#
# Далее кинуть скрипт и проверить что он работает командой --help
#
# python3 pymongoinfo.py --help
#
# Принимает он только dsn к монге
#
# python3 pymongoinfo.py mongodb://root:secret-pass@hostname:port
#
# Данные сохранит в файлик output.json в папке в которой его запускали
import json
from pathlib import Path
import click
from pymongo import MongoClient
from pymongo.errors import OperationFailure
def _get_databases(
client,
):
databases = client.list_databases()
return {database['name']: database for database in databases}
def _get_collections(
database,
):
return tuple(database.list_collection_names())
def _get_indexes(
collection,
):
return tuple(collection.list_indexes())
@click.command()
@click.argument(
'mongodb_dsn',
required=True,
)
@click.option(
'--output', '-o',
required=True,
default='output.json',
show_default=True,
type=click.Path(
resolve_path=True,
dir_okay=False,
path_type=Path,
),
)
def main(
mongodb_dsn: str,
output: Path,
):
"""
USAGE EXAMPLE:
python3 pymongoinfo.py mongodb://root:secret-pass@hostname:port
"""
client = MongoClient(mongodb_dsn)
click.echo('Read databases info')
result_dict = _get_databases(client)
for db, value in result_dict.items():
click.echo('Read collections info for ' + db)
try:
cols = _get_collections(client[db])
value['collection'] = {col: {'indexes': _get_indexes(client[db][col])} for col in cols}
except OperationFailure:
click.echo('Skip ' + db)
result_str = json.dumps(result_dict, indent=2, sort_keys=True)
click.echo('Write results to ' + str(output))
with open(output, 'w') as fh:
fh.write(result_str)
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment