Discord.py bot reading mariaDB and creating embeds

Question

My first real Python project. I have a web scraping PowerShell utility that scrapes Kijiji (Canadian Craigslist) for listings and loads them into MariaDB. What the bot does is periodically check the database for listing flagged as new. Creates embeds from the listing data and post them to an appropriate discord channel (also removes the new flag). I really like the import system so I tried to break up the project into logical chunks. The bot settings are driven by JSON. A few bot commands were added in here as well.

This is the main py file. It is expecting a config file called bot_cfg.json to be in the same directory from where it was executed. For me that is fine. It would be nicer to accept the path as input but I prefer the assumption for now.

kjiji-bot.py

# Kijiji Bot
# Uses Python 3.6.5

# Discord specific imports
import discord
from discord.ext import commands
from discord.ext.commands import Bot
# Miscellaneous imports
import asyncio
import logging
import os
from pathlib import Path
from random import randint
from datetime import datetime
# Database
from sqlalchemy import create_engine, and_
from sqlalchemy.orm import Session
from sqlalchemy.orm.exc import NoResultFound
# Custom Class
from listing import Listing, Base
from botconfig import BotConfig

# Set up Discord logging
logger = logging.getLogger('discord')
logger.setLevel(logging.DEBUG)
handler = logging.FileHandler(filename='discord.log', encoding='utf-8', mode='w')
handler.setFormatter(logging.Formatter('%(asctime)s:%(levelname)s:%(name)s: %(message)s'))
logger.addHandler(handler)

# Configuration file name. To be located in same directory as script
config_file_name = "bot_cfg.json"

class KijijiListing(object):
    '''The basic Kijiji Listing information'''

    def __init__(self, dictionary):
        self.url = dictionary['absoluteurl']
        self.imageurl = dictionary['imageurl']
        self.id = dictionary['id']
        self.posted = dictionary['postedasdate']
        self.title = dictionary['title']
        self.description = dictionary['shortdescription']
        self.location = dictionary['location']
        self.price = dictionary['price']
        self.thumbnail = 'https://www.shareicon.net/data/128x128/2016/08/18/810389_strategy_512x512.png'

    def __str__(self):
        return 'Title: {}\nDescription: {}\nPrice: {}\nURL: {}'.format(
            self.title, self.description, self.price, self.url
        )

    def to_embed(self):
        '''Created a discord embed from this instances properties'''
        listing_embed = discord.Embed(
            title=self.title, description=self.description, color=discord.Colour(randint(0, 16777215)),
            url=self.url)
        listing_embed.add_field(name='Location', value=self.location, inline=True)
        listing_embed.add_field(name='Price', value=self.price, inline=True)
        listing_embed.set_image(url=self.imageurl)
        listing_embed.set_thumbnail(
            url=self.thumbnail)
        listing_embed.set_footer(text='Listed: {}'.format(self.posted))
        return listing_embed

# Scripts running location. Only set if called via python.exe
__location__ = os.path.realpath(
    # From https://docs.python.org/3/library/os.path.html
    # If a component is an absolute path, all previous components
    # are thrown away and joining continues from the absolute path component.
    os.path.join(os.getcwd(), os.path.dirname(__file__)))

# Load Configuration File
config_file_path = Path(os.path.join(__location__, config_file_name))

# Read in configuration file.
if(config_file_path.is_file()):
    print("Configuration found in: {}".format(config_file_path))

    # Initiate the bot config object from file
    bot_config = BotConfig(config_file_path)
    print(str(bot_config))
else:
    print("The configuration file {} does not exist".format(config_file_path))

# Initialize the bot
bot = Bot(command_prefix=bot_config.command_prefix)

# Prep SQLAlchemy
engine = create_engine(bot_config.db_url, pool_recycle=3600)
session = Session(bind=engine)
Base.metadata.create_all(engine)

@bot.event
async def on_ready():
    '''Event for when the bot is ready to start working'''
    print("Ready when you are")
    print("I am running on " + bot.user.name)
    print("With the id " + bot.user.id)
    await bot.change_presence(game=discord.Game(name=bot_config.randompresence()))

@bot.command(pass_context=True)
async def ping(context, *args):
    '''Verifcation that the bot is running and working.'''
    await bot.say(":eight_spoked_asterisk: I'm here {}".format(
        context.message.author))
    # Remove the message that triggered this command
    await bot.delete_message(context.message)
    print("{} has pinged".format(context.message.author))

@bot.command(pass_context=True)
@commands.has_role('admin')
async def shutdown(context):
    '''Command to shut the bot down'''
    # Remove the message that triggered this command
    await bot.delete_message(context.message)
    await bot.logout()

@bot.command(pass_context=True)
async def status(context):
    ''' Reports pertinent bot statistics as an embed'''
    status_embed = discord.Embed(
        title="Kijiji Bot Status",
        description="Quick snapshot of what is going on with the bot",
        color=discord.Colour(randint(0, 16777215))
    ).add_field(
        name='Last DB Check',
        value=bot_config.last_searched
    )

    await bot.send_message(destination=context.message.channel, embed=status_embed)
    # Remove the message that triggered this command
    await bot.delete_message(context.message)

@bot.command(pass_context=True, aliases=['np'])
async def newpresence(context):
    '''Change the bot presence to another from from config'''

    # Get the current status of the bot so we can omit that from the choices.
    current_game = ([single_member.game.name for single_member in bot.get_all_members() if single_member.id == bot.user.id])[0]

    # Check to see if we have multiple options to choose from
    if len(bot_config.presence) > 1:
        # Same one could possibly show.
        await bot.change_presence(game=discord.Game(name=bot_config.randompresence(current_game)))
    else:
        await bot.say('I only have one presence.')

    # Remove the message that triggered this command
    await bot.delete_message(context.message)

@bot.command(pass_context=True, aliases=['gl'])
async def getlisting(context, id):
    '''Get a listing from the database matching the id passed'''
    try:
        single_listing = session.query(Listing).filter(Listing.id == id).first()
    except NoResultFound as e:
        print(e)
        await bot.say("No listings available")
        # Deal with that as wells

    if(single_listing):
        # print("channel:", bot_config.search[0].posting_channel.name, bot_config.search[0].posting_channel.id)
        await bot.send_message(destination=bot_config.search[0].posting_channel, embed=single_listing.to_embed())

    # Remove the message that triggered this command
    await bot.delete_message(context.message)


async def listing_watcher():
    ''' This is the looping task that will scan the database for new listings and post them to their appropriate channel'''
    await bot.wait_until_ready()

    while not bot.is_closed:
        # Process each search individually
        for single_search in bot_config.search:
            # Attempt to get new listings up to a certain number
            try:
                new_listings = session.query(Listing).filter(and_(Listing.new == 1, Listing.searchurlid.in_(single_search.search_indecies))).limit(bot_config.posting_limit)
            except NoResultFound as e:
                await bot.say("No listings available")

            if(new_listings):
                for new_listing in new_listings:
                    await bot.send_message(destination=single_search.posting_channel, embed=new_listing.to_embed())
                    # Flag the listing as old
                    new_listing.new = 0

                session.commit()

            # Update the last search value in config. Used in status command
            bot_config.last_searched = datetime.now()

            # Breather between search configs
            await asyncio.sleep(1)

        # task runs every 60 seconds
        await asyncio.sleep(60)

# Run the bot with the supplied token
print('Discord.py version:', discord.__version__)

# Start the database monitoring task
bot.loop.create_task(listing_watcher())
bot.run(bot_config.token)
bot.close()

botconfig.py

from json import load
from random import choice
from validators import url
from discord import Object

class SearchConfig(object):
    '''Coordinates the pulling of listings from the database and posting them'''
    __slots__ = 'id', 'search_indecies', 'posting_channel', 'thumbnail'

    def __init__(self, dictionary):

        if 'id' in dictionary.keys():
            self.id = dictionary['id']
        else:
            raise ValueError('Parameter "id" is required')

        if 'search_indecies' in dictionary.keys():
            self.search_indecies = [item for item in dictionary['search_indecies'] if isinstance(item, int)]

            # Verify that search contains at least one item
            if self.search_indecies.count == 0:
                raise ValueError('Parameter "search" contains no integers')
        else:
            raise ValueError('Parameter "search_indecies" is required')

        if 'posting_channel' in dictionary.keys():
            self.posting_channel = Object(id=dictionary['posting_channel'])
        else:
            raise ValueError('Parameter "posting_channel" is required')

        if 'thumbnail' in dictionary.keys():
            if url(dictionary['thumbnail']):
                self.thumbnail = dictionary['thumbnail']
            else:
                print(f"Thumbnail for {self.id} failed url validation")

    def __str__(self):
        return 'Id: {}\nSearch Indecies: {}\nPosting Channel: {}'.format(
            self.id,
            ", ".join([str(x) for x in self.search_indecies]),
            self.posting_channel
        )


class BotConfig(object):
    ''' Contains all import configuration used for the bot'''
    __slots__ = 'command_prefix', 'token', 'search', 'presence', 'db_url', 'posting_limit', 'last_searched'

    def __init__(self, path):
        '''Using the file path of the config file import and scrub settings '''

        # Set bot defaults where applicable
        defaults = dict(command_prefix='#', presence='hard to get', posting_limit=3)

        # Load from file
        with open(path) as json_file:
            config_options = load(json_file)

        # Check for the required token property
        if 'token' in config_options.keys():
            self.token = config_options['token']
        else:
            raise ValueError('Parameter "token" is required.')

        # Check for the optional posting_limit
        if 'posting_limit' in config_options.keys():
            self.posting_limit = config_options['posting_limit']
        else:
            self.posting_limit = defaults['posting_limit']

        # Get the required database url
        if 'db_url' in config_options.keys():
            self.db_url = config_options['db_url']
        else:
            raise ValueError('Parameter "db_url" is required.')

        # Check for the required search object property
        self.search = []
        if 'search' in config_options.keys():
            for search_config in config_options['search']:
                self.search.append(SearchConfig(dictionary=search_config))
        else:
            raise ValueError('At least one "search" is required.')

        # Set the command prefix from config if possible
        if "command_prefix" in config_options.keys():
            self.command_prefix = config_options["command_prefix"]
        else:
            self.command_prefix = defaults['command_prefix']

        # Load presences if any. Append default just in case
        self.presence = []
        if 'presence' in config_options.keys():
            self.presence = config_options['presence']

        if self.presence.count == 0:
            self.presence.append(defaults['presence'])

    def randompresence(self, *args):
        # Get a random presence from list
        return choice([option for option in self.presence if option not in args])

    def __str__(self):
        return 'Command Prefix: {} \nSearch: {} \nToken: {} \nPresence: {}'.format(
            self.command_prefix,
            "\n".join(str(x) for x in self.search),
            self.token,
            self.presence
        )

bot_cfg.json

{
    "token": "NDMwODA1O-FAKETOKEN-Mvsz-es",
    "command_prefix" : "?",
    "db_url": "mysql+pymysql://kijiji:zA8oCA1I88Wo@localhost/kijiji",
    "search": [
        {
            "id": "boardgames",
            "search_indecies" : [2],
            "posting_channel": "436277304260558848",
            "thumbnail" : "https://www.shareicon.net/data/128x128/2016/08/18/810389_strategy_512x512.png"
        }   
    ],
    "presence": [
        "with fire",
        "at full blast",
        "for keeps",
        "a trick on you",
        "Cupid",
        "Knify Spoony",
        "hooky",
        "it cool",
        "for all the marbles",
        "the market",
        "the trump card",
        "possum",
        "with loaded dice"
    ]
}

listing.py

from discord import Embed, Colour
from json import loads
from random import randint
from sqlalchemy import Column, String, Integer, Boolean, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

# https://auth0.com/blog/sqlalchemy-orm-tutorial-for-python-developers/
class SearchURL(Base):
    __tablename__ = 'searchurls'
    urlid = Column(Integer, primary_key=True)
    url = Column(String)
    inserted = Column(DateTime)

    def __str__(self):
        return ('urlid:' + str(self.urlid) + '\n'
                'url:' + self.url + '\n'
                'inserted:' + str(self.inserted)
                )

class Listing(Base):
    __tablename__ = 'listings'

    id = Column(Integer, primary_key=True)
    url = Column(String)
    price = Column(String)
    title = Column(String)
    distance = Column(String)
    location = Column(String)
    posted = Column(DateTime)
    shortdescription = Column(String)
    lastsearched = Column(DateTime)
    searchurlid = Column(Integer, ForeignKey('searchurls.urlid'))
    imageurl = Column(String)
    discovered = Column(Integer)
    new = Column(Boolean)
    changes = Column(String)
    searchurl = relationship('SearchURL')

    def __str__(self):
        return ('id: ' + str(self.id) + "\n" +
                'url: ' + self.url + "\n" +
                'price: ' + self.price + "\n" +
                'title: ' + self.title + "\n" +
                'distance: ' + self.distance + "\n" +
                'location: ' + self.location + "\n" +
                'posted: ' + str(self.posted) + "\n" +
                'shortdescription: ' + self.shortdescription + "\n" +
                'lastsearched: ' + str(self.lastsearched) + "\n" +
                'searchurlid: ' + str(self.searchurlid) + "\n" +
                'imageurl: ' + self.imageurl + "\n" +
                'discovered: ' + str(self.discovered) + "\n" +
                'new: ' + str(self.new) + "\n" +
                'changes: ' + self.changes + "\n" +
                'searchurl: ' + str(self.searchurl)
                )

    def changes_to_string(self):
        '''Take the json string changes and convert it to formatted string for reading in embed'''
        changes_markdown = ''
        if(self.changes):
            for change in loads(self.changes):
                changes_markdown += f'_{str(change["Property"])}:_ {str(change["Findings"])}\n'
        return changes_markdown

    def to_embed(self, **kwargs):
        '''Created a discord embed from this instances properties'''
        # If this listing has been discovered before then it is possible there are changes
        # that should be shown in the message as well.
        if(self.discovered > 0):
            listing_description = (
                f'{self.shortdescription}\n\n'
                f'This listing has been found {self.discovered} time(s) before.\n'
            )
            if self.changes:
                listing_description += (
                    f'**The following differences from the previous listing were identified**\n'
                    f'{self.changes_to_string()}'
                )
        else:
            listing_description = self.shortdescription

        listing_as_embed = Embed(
            title=self.title, description=listing_description, color=Colour(randint(0, 16777215)),
            url=f'https://www.kijiji.ca{self.url}')
        listing_as_embed.add_field(name='Location', value=self.location, inline=True)
        listing_as_embed.add_field(name='Price', value=self.price, inline=True)
        # This url might contain a tilde which Discord will have an issue with
        # Replace the tilde with a URL encoded version
        listing_as_embed.set_image(url=self.imageurl.replace('~', '%7E'))
        listing_as_embed.set_footer(text='Listed: {}'.format(self.posted))
        if 'thumbnail' in kwargs:
            listing_as_embed.set_thumbnail(url=kwargs.get('thumbnail'))

        return listing_as_embed

Since this is my first project I know there are areas of improvement. This was my first look at asynchronous programming, discord.py, SQLAlchemy, and most of the other concepts used here. Those and how I am dealing with JSON would be appreciated areas of focus.

I used Anaconda in Sublime Text so most of the PEP8 stuff should be good. I did disable E501 (Line length), E302(double blank after imports) checks as I found that mostly annoying.

Environment Details

• Python 3.6.5
• Discord.py 0.16.12
• MariaDB 10.2.12-MariaDB

Answer 1

I haven't used Discord or MariaDB, so I'll let others take a look at those areas.

---

if(config_file_path.is_file()):
    bot_config = BotConfig(config_file_path)
else:
    print("The configuration file {} does not exist".format(config_file_path))

bot = Bot(command_prefix=bot_config.command_prefix)

You don't need ()'s in an if cond: statement.

If the file does not exist, you print an error message and muddle on, but the very next statement is going to raise the exception "Name error: name 'bot_config' is not defined." You should raise an exception in the else: clause, not print a message and muddle on.

---

current_game = ([single_member.game.name for single_member in bot.get_all_members() if single_member.id == bot.user.id])[0]

Here you are using list comprehension to construct an entire list, and the get only the first item in the list. The construction of the intermediate list is busy work which is unneeded. Instead, consider the next() function, which returns the next value in an iterable ... which will be just the first value if not used in a loop.

current_game = next(single_member.game.name
                    for single_member in bot.get_all_members()
                    if single_member.id == bot.user.id)

---

try:
    single_listing = session.query(Listing).filter(Listing.id == id).first()
except NoResultFound as e:
    print(e)
    await bot.say("No listings available")

if(single_listing):

The error recovery, is broken. If NoResultFound is raised, then single_listing won't have been created, and the if single_listing: statement will raise a NameError.

Ditto for new_listings in later code.

---

Configuration

defaults = dict(command_prefix='#', presence='hard to get', posting_limit=3)

#...

if 'posting_limit' in config_options.keys():
    self.posting_limit = config_options['posting_limit']
else:
    self.posting_limit = defaults['posting_limit']

The string 'posting_limit' is repeated 3 times here, plus as a key to the defaults dictionary. Plenty of chances for a typo to go unnoticed. You could change this to:

self.posting_limit = config_options.get('posting_limit', 3)  

But this still doesn't protect against a "spelling" error in the configuration file. If the config file has

"post_limit": 5,

The default will be used, and no warning/error will be given for an unknown configuration item.

My preferred initialization method would be:

class BotConfig:
    '''Contain all import configuration for the bot'''

    def __init__(self, *, token, db_url, search, command_prefix='#',
                 presence=['hard to get'], posting_limit=3):
        '''Set and scrub settings'''

        self.posting_limit = posting_limit
        # ... etc ...

The default values are listed right in the constructor declaration. The keyword arguments have to be spelt out exactly. Keywords without defaults must be provided. Unrecognized keywords are forbidden. (Note: The * is a flag that requires all remaining arguments have to be keyword arguments.)

Except you want to read in the configuration from a json file. This isn't a problem. We just add a factory method for this additional constructor.

    @classmethod
    def from_json_config(cls, path):
        '''Using the file path of the config file import and scrub settings'''

        with open(path) as json_file:
            config_options = load(json_file):
            return cls(**config_options)

The **config_options explodes the config_options dictionary into a list of key=value pairs. If the config file contains:

 "command_prefix": "?",
 "post_limit": 5,

then config_options would be {'command_prefix': '?', 'post_limit': 5), and the cls(**config_options) would expand to BotConfig(command_prefix='?', post_limit=5, posting_limit=3, presence=['hard to get']), and a "TypeError: BotConfig() got an unexpected keyword argument 'post_limit'" exception would be raised (or missing token or db_url or search keyword, because I was lazy).

Use a similar setup for SearchConfig constructor:

self.search = []
for search_config in search:
    self.search.append(SearchConfig(**search_config))
if len(self.search) == 0:
    raise ValueError('At least one "search" is required.')

---

", ".join([str(x) for x in self.search_indecies])

Pointless creation of a list. Use:

", ".join(str(x) for x in self.search_indecies)

---

Again, pointless creation of a list:

def randompresence(self, *args):
    # Get a random presence from list
    return choice([option for option in self.presence if option not in args])

And missing a '''docstring''', so I can't tell if the condition is right. Is randompresence(...) suppose to choose from the list of arguments provided (as long as they are a valid option in self.presence), or excluding an option given in the argument list?

Or maybe you wanted choose from the entire list, if no argument is given:

if args:
    return choice(option for option in self.presence if option in args)
else:
    return choice(self.presence)

---

Lots of unnecessary string concatenation.

def __str__(self):
    return ('urlid:' + str(self.urlid) + '\n'
            'url:' + self.url + '\n'
            'inserted:' + str(self.inserted)
            )

Could use a format string instead.

def __str__(self):
    return (f'urlid:{self.urlid}\n'
            f'url:{self.url}\n'
            f'inserted:{self.inserted}'
            )

Or, if you get tired of typing in self. for all the fields (I'm looking at Listing.__str__), {key} will be replaced with the keyword argument given in the format(...) arguments, and the splat operator turns the self.__dict__ into key=value pairs, for use as the keyword arguments for format(...):

def __str__(self):
    return ('urlid:{urlid}\n'
            'url:{url}\n'
            'inserted:{inserted}'
            ).format(**self.__dict__)

For __slots__ based classes, you could use:

def __str__(self):
    return ('your format string here'
            ).format(**{key: getattr(self, key) for key in self.__slots__})

Of course, if you are always displaying as multiple lines of keyword:{keyword}, and don't care about the order, you could use:

def __str__(self):
   return "\n".join(f"{key}:{val}" for key, val in vars(self).items())

And for __slots__ based classes, you could use:

def __str__(self):
    return "\n".join(f"{key}:{getattr(self, key)}" for key in self.__slots__)

---

Error: comparison between method and integer:

if self.presence.count == 0:
    self.presence.append(defaults['presence'])

self.presence is a list [], which has a method .count, which counts the occurrence of an item within the list. It is not the length of the list. Use:

if len(self.presence) == 0:

or, since an empty list is falsy ...

if not self.presence: