flake8 cleanup

main
lza_menace 3 years ago
parent 5b7641d16d
commit 9561eb668b

@ -5,7 +5,7 @@ from suchwowx.factory import db
bp = Blueprint('cli', 'cli', cli_group=None)
@bp.cli.command('init')
def init():
import suchwowx.models
db.create_all()

@ -7,6 +7,7 @@ from suchwowx.factory import db
bp = Blueprint('mod', 'mod')
@bp.cli.command('list')
def list():
"""
@ -15,6 +16,7 @@ def list():
for mod in Moderator.query.all():
click.echo(mod.user.handle)
@bp.cli.command('add')
@click.argument('moderator_handle')
def add(moderator_handle):
@ -32,6 +34,7 @@ def add(moderator_handle):
else:
click.echo('[.] That is not a valid user.')
@bp.cli.command('remove')
@click.argument('moderator_handle')
def remove(moderator_handle):
@ -44,8 +47,8 @@ def remove(moderator_handle):
if mod:
db.session.delete(mod)
db.session.commit()
click.echo(f'[-] Removed moderator status from `{moderator_handle}`')
click.echo(f'[-] Removed moderator status from `{moderator_handle}`') # noqa
else:
click.echo(f'[.] That user is not a moderator.')
click.echo('[.] That user is not a moderator.')
else:
click.echo('[.] That is not a valid user.')

@ -4,11 +4,10 @@ from dotenv import load_dotenv
load_dotenv()
# App
SECRET_KEY = getenv('SECRET_KEY', 'yyyyyyyyyyyyy') # whatever you want it to be
DATA_FOLDER = getenv('DATA_FOLDER', '/path/to/uploads') # some stable storage path
SERVER_NAME = getenv('SERVER_NAME', '127.0.0.1:5000') # name of your DNS resolvable site (.com)
IPFS_SERVER = getenv('IPFS_SERVER', 'http://127.0.0.1:8080') # ip/endpoint of ipfs explorer
MOD_MODE = getenv('MOD_MODE', 1) # mod mode enabled by default (enforce queue)
SECRET_KEY = getenv('SECRET_KEY', 'yyyyyyyyyyyyy')
DATA_FOLDER = getenv('DATA_FOLDER', '/path/to/uploads')
SERVER_NAME = getenv('SERVER_NAME', '127.0.0.1:5000')
IPFS_SERVER = getenv('IPFS_SERVER', 'http://127.0.0.1:8080')
# Cache
CACHE_HOST = getenv('CACHE_HOST', 'localhost')
@ -21,7 +20,7 @@ MAX_CONTENT_LENGTH = 32 * 1024 * 1024
TEMPLATES_AUTO_RELOAD = getenv('TEMPLATES_AUTO_RELOAD', True)
# Contract
CONTRACT_ADDRESS = '0xBAb68B24068D21Fa862908818054c7e4d921db5A' # rinkeby ETH
CONTRACT_ADDRESS = '0xBAb68B24068D21Fa862908818054c7e4d921db5A' # rinkeby ETH
CONTRACT_ABI = [
{
"inputs": [],

@ -10,15 +10,16 @@ from suchwowx import config
db = SQLAlchemy()
w3 = Web3(Web3.HTTPProvider('http://127.0.0.1:9650')) # todo
w3 = Web3(Web3.HTTPProvider('http://127.0.0.1:9650')) # noqa
def setup_db(app: Flask, db:SQLAlchemy=db):
def setup_db(app: Flask, db: SQLAlchemy = db):
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{config.DATA_FOLDER}/sqlite.db' # noqa
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)
return db
def create_app_huey():
app = Flask(__name__)
db = SQLAlchemy()
@ -26,6 +27,7 @@ def create_app_huey():
setup_db(app, db)
return app
def create_app():
app = Flask(__name__)
app.config.from_envvar('FLASK_SECRETS')
@ -42,7 +44,7 @@ def create_app():
return user
with app.app_context():
from suchwowx import filters, cli
from suchwowx import filters
from suchwowx.routes import api, meme, meta, user
from suchwowx.cli import mod, cli
app.register_blueprint(filters.bp)

@ -4,12 +4,14 @@ from arrow import get as arrow_get
bp = Blueprint('filters', 'filters')
@bp.app_template_filter('shorten_address')
def shorten_address(a):
_p = a[0:6]
_s = a[-4:]
return f'{_p}...{_s}'
@bp.app_template_filter('humanize')
def humanize(d):
return arrow_get(d).humanize()

@ -2,7 +2,7 @@ import ipfsApi
from eth_account.messages import encode_defunct
from suchwowx.models import Meme
from suchwowx.factory import w3, db
from suchwowx.factory import w3
from suchwowx import config
@ -14,6 +14,7 @@ def verify_signature(message, signature, public_address):
else:
return False
def upload_to_ipfs(meme_id: str):
meme = Meme.query.get(meme_id)
if not meme:
@ -37,4 +38,5 @@ def upload_to_ipfs(meme_id: str):
print(f'[+] Uploaded metadata to IPFS: {meta_hash}')
return (meta_hash, artwork_hash)
except Exception as e:
print(f'[!] Error: {e}')
return False

@ -1,6 +1,7 @@
from uuid import uuid4
from datetime import datetime
from flask import url_for
from flask_login import login_user
from suchwowx.factory import db
@ -10,6 +11,7 @@ from suchwowx import config
def rand_id():
return uuid4().hex
class Moderator(db.Model):
__tablename__ = 'moderators'
@ -20,6 +22,7 @@ class Moderator(db.Model):
def __rep__(self):
return self.user.handle
class User(db.Model):
__tablename__ = 'users'
@ -65,7 +68,7 @@ class User(db.Model):
def get_profile_image(self, full=True):
if self.profile_image:
if full:
return url_for('meta.uploaded_file', filename=self.profile_image)
return url_for('meta.uploaded_file', filename=self.profile_image) # noqa
else:
return self.profile_image
else:

@ -1,6 +1,6 @@
from secrets import token_urlsafe
from flask import Blueprint, request, jsonify, url_for
from flask import Blueprint, request, jsonify
from flask_login import current_user
from suchwowx.factory import db
@ -10,6 +10,7 @@ from suchwowx.models import User
bp = Blueprint('api', 'api', url_prefix='/api/v1')
@bp.route('/user_exists')
def user_exists():
"""
@ -39,6 +40,7 @@ def user_exists():
'success': True
})
@bp.route('/authenticate/metamask', methods=['POST'])
def authenticate_metamask():
"""
@ -63,7 +65,7 @@ def authenticate_metamask():
if _u:
if data['message'].endswith(_u.nonce):
if verify_signature(data['message'], data['signed_data'], data['public_address']):
if verify_signature(data['message'], data['signed_data'], data['public_address']): # noqa
_u.login()
return jsonify({
'success': True,

@ -1,15 +1,13 @@
from os import path, remove
from secrets import token_urlsafe
from json import loads, dumps
import ipfsApi
from flask import Blueprint, render_template, request, current_app
from flask import send_from_directory, redirect, flash, url_for, jsonify
from flask_login import logout_user, current_user, login_user
from requests.exceptions import HTTPError
from flask import Blueprint, render_template, request
from flask import redirect, flash, url_for
from flask_login import current_user
from web3 import Web3
from suchwowx.models import Meme, User
from suchwowx.models import Meme
from suchwowx.helpers import upload_to_ipfs
from suchwowx.factory import db
from suchwowx import config
@ -17,9 +15,12 @@ from suchwowx import config
bp = Blueprint('meme', 'meme')
@bp.route('/')
def index():
memes = Meme.query.filter(Meme.meta_ipfs_hash != None).order_by(Meme.create_date.desc())
memes = Meme.query.filter(
Meme.meta_ipfs_hash != None
).order_by(Meme.create_date.desc())
w3 = Web3(Web3.HTTPProvider('http://127.0.0.1:9650'))
contract_address = w3.toChecksumAddress(config.CONTRACT_ADDRESS)
contract_abi = config.CONTRACT_ABI
@ -30,6 +31,7 @@ def index():
# total_supply = contract.functions.totalSupply().call()
return render_template('index.html', memes=memes, contract=contract)
@bp.route('/mod')
def mod():
if not current_user.is_moderator():
@ -40,6 +42,7 @@ def mod():
).order_by(Meme.create_date.asc())
return render_template('index.html', memes=memes)
@bp.route('/publish', methods=['GET', 'POST'])
def publish():
if not current_user.is_authenticated:
@ -83,9 +86,9 @@ def publish():
meme.meta_ipfs_hash = res[0]
meme.meme_ipfs_hash = res[1]
db.session.commit()
flash('Published new meme to local database and IPFS.', 'success')
flash('Published new meme to local database and IPFS.', 'success') # noqa
else:
flash('Published new meme to database for review by moderators.', 'success')
flash('Published new meme to database for review by moderators.', 'success') # noqa
return redirect(url_for('meme.index'))
except ConnectionError:
flash('[!] Unable to connect to local ipfs', 'error')
@ -96,6 +99,7 @@ def publish():
meme=meme
)
@bp.route('/meme/<meme_id>')
def show(meme_id):
meme = Meme.query.filter(Meme.id == meme_id).first()
@ -109,6 +113,7 @@ def show(meme_id):
return redirect(url_for('meme.index'))
return render_template('meme.html', meme=meme)
@bp.route('/meme/<meme_id>/<action>')
def approve(meme_id, action):
if not current_user.is_authenticated:
@ -129,10 +134,14 @@ def approve(meme_id, action):
if not res:
flash('Unable to post to IPFS, daemon may be offline.', 'error')
return redirect(url_for('meme.show', meme_id=meme.id))
existing_meta_ipfs = Meme.query.filter(Meme.meta_ipfs_hash == res[0]).first()
existing_meme_ipfs = Meme.query.filter(Meme.meme_ipfs_hash == res[1]).first()
existing_meta_ipfs = Meme.query.filter(
Meme.meta_ipfs_hash == res[0]
).first()
existing_meme_ipfs = Meme.query.filter(
Meme.meme_ipfs_hash == res[1]
).first()
if existing_meta_ipfs or existing_meme_ipfs:
flash('Cannot use an existing IPFS hash for either metadata or memes.', 'warning')
flash('Cannot use an existing IPFS hash for either metadata or memes.', 'warning') # noqa
return redirect(url_for('meme.show', meme_id=meme.id))
meme.meta_ipfs_hash = res[0]
meme.meme_ipfs_hash = res[1]

@ -7,6 +7,7 @@ from suchwowx import config
bp = Blueprint('meta', 'meta')
@bp.route('/uploads/<path:filename>')
def uploaded_file(filename):
"""
@ -14,10 +15,12 @@ def uploaded_file(filename):
"""
return send_from_directory(f'{config.DATA_FOLDER}/uploads', filename)
@bp.route('/about')
def about():
return render_template('about.html')
@bp.route('/disconnect')
def disconnect():
logout_user()

@ -1,13 +1,12 @@
from flask import Blueprint, render_template
from flask import redirect, url_for
from flask_login import logout_user
from suchwowx.models import User
from suchwowx import config
bp = Blueprint('user', 'user')
@bp.route('/user/<handle>')
def show(handle):
user = User.query.filter(User.handle == handle).first()

Loading…
Cancel
Save