Skip to content

Instantly share code, notes, and snippets.

@arindampradhan
Created August 5, 2016 12:23
Show Gist options
  • Save arindampradhan/270c39b2df65ebe95999bfaeb5a5beb3 to your computer and use it in GitHub Desktop.
Save arindampradhan/270c39b2df65ebe95999bfaeb5a5beb3 to your computer and use it in GitHub Desktop.
Config and manage script for flask | python
import os
basedir = os.path.abspath(os.path.dirname(__file__))
class Config(object):
DEBUG = False
TESTING = False
CSRF_ENABLED = True
SECRET_KEY = '' # os.urandom(24)
class ProductionConfig(Config):
DEBUG = False
class StagingConfig(Config):
DEVELOPMENT = True
DEBUG = True
class DevelopmentConfig(Config):
DEVELOPMENT = True
DEBUG = True
class TestingConfig(Config):
TESTING = True
class DB(object):
db = ''
host = 'localhost'
port = None
username = None
password = None
class MongoDB(DB):
"""URL FORMAT - mongodb://username:password@host:port/database?options"""
db = ''
host = ''
port = 27100
username = ''
password = ''
url = "mongodb://{}:{}@{}:{}/{}".format(
username, password, host, port, db)
class RedisDB(DB):
"""URL FORMAT - redis://:password@hostname:port/db_number"""
host = ''
port = 18772
username = ''
password = ''
url = "redis://:{}@{}:{}/{}".format(
password, host, port, 0)
from flask_script import Manager
from logging import FileHandler, WARNING
from config import MongoDB
from pymongo import MongoClient
from bson.json_util import dumps
from app import app
# Manager for app
manager = Manager(app)
@manager.command
def log(task):
"""Enable and disable logging in flask."""
if task == 'Enable':
if not app.debug:
file_handler = FileHandler('errorlog.txt')
file_handler.setLevel(WARNING)
app.logger.addHandler(file_handler)
if task == 'disable':
# TODO
pass
# Mongo Commands
@manager.command
def rmcollection(dbname, collection):
"""Remove a collection from a database in MongoDB."""
url = MongoDB.url
client = MongoClient(url)
client[dbname][collection].drop()
print "Removed {} in {} database".format(dbname, collection)
@manager.command
def mongotocsv(collection, OutputFile):
"""Store Mongo data to csv file."""
if not OutputFile.endswith('.csv'):
print "Please mention a csv file in OutputFile!"
print "Storing the result in result.csv"
OutputFile = "result.csv"
import subprocess
from config import MongoDB
dbnameORuri = MongoDB.url.split('/')[-1]
subprocess.call(
['python', './Helpers/mongoexportcsv.py', dbnameORuri, collection, OutputFile])
print "Importing {1} collection data from {0} to the file path {2}".format(dbnameORuri, collection, OutputFile)
@manager.command
def mongotojson(collection, OutputFile):
"""Stores Mongo data as json file."""
if not OutputFile.endswith('.json'):
print "Please mention a json file in OutputFile!"
print "Storing the result in result.json"
OutputFile = "result.json"
url = MongoDB.url
client = MongoClient(url)
dbname = MongoDB.url.split('/')[-1]
bsons = client[dbname][collection].find({}) # find all
with open(OutputFile, "w+") as f:
for i in bsons:
json_string = dumps(i) + "\n"
f.write(json_string)
print "Importing {1} collection data from {0} to the file path {2}".format(url, collection, OutputFile)
if __name__ == "__main__":
manager.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment