Created
December 7, 2022 08:04
-
-
Save shpaker/ad563fbaecaf2b07817013449a1a0994 to your computer and use it in GitHub Desktop.
Скриптик для сбора базовой инфы о том, что лежит в монге
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# В общем, на хосте где есть третий питон и доступ к этой монге надо поставить пакеты 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