Edit on GitHub

tickets_plus.api.handlers

Application Interface Handlers.

This module contains the handlers for the application interface. As it stands, these are not for use by the end user, but rather can be used by the main bot to deliver ticket data to our app. Not to be used directly, but rather through the routes' module.

Typical usage example:

from tickets_plus.api import handlers

...

View Source

  1"""Application Interface Handlers.
  2
  3This module contains the handlers for the application interface.
  4As it stands, these are not for use by the end user, but rather
  5can be used by the main bot to deliver ticket data to our app.
  6Not to be used directly, but rather through the routes' module.
  7
  8Typical usage example:
  9    ```py
 10    from tickets_plus.api import handlers
 11
 12    ...
 13    ```
 14"""
 15# License: EPL-2.0
 16# SPDX-License-Identifier: EPL-2.0
 17# Copyright (c) 2021-present The Tickets+ Contributors
 18# This Source Code may also be made available under the following
 19# Secondary Licenses when the conditions for such availability set forth
 20# in the Eclipse Public License, v. 2.0 are satisfied: GPL-3.0-only OR
 21# If later approved by the Initial Contributor, GPL-3.0-or-later.
 22
 23import json
 24
 25import discord
 26from sqlalchemy import orm
 27from tornado import web
 28
 29from tickets_plus import bot
 30from tickets_plus.cogs import events
 31from tickets_plus.database import models
 32
 33
 34class BotHandler(web.RequestHandler):
 35    """Handler for the bot to send data to the app.
 36
 37    This handler is used to receive data into the bot
 38    """
 39
 40    def initialize(self, bot_instance: bot.TicketsPlusBot) -> None:
 41        """Initialize the handler.
 42
 43        Initialize the handler with the bot object.
 44
 45        Args:
 46            bot_instance: The bot object.
 47        """
 48        self._bt = bot_instance
 49        self.SUPPORTED_METHODS = ("POST",)  # pylint: disable=invalid-name
 50
 51    def set_default_headers(self) -> None:
 52        """Sets the return type to JSON.
 53
 54        Not striclty necessary, but it's good practice.
 55        """
 56        self.set_header("Content-Type", "application/json")
 57        # pylint: disable=line-too-long # skipcq: FLK-E501
 58        self.set_header("Strict-Transport-Security", "max-age=63072000; includeSubDomains; preload")
 59        # pylint: disable=line-too-long # skipcq: FLK-E501
 60        self.set_header(
 61            "Content-Security-Policy", "default-src 'none'; script-src 'none'; object-src 'none'; "
 62            "style-src 'none'; img-src 'none'; font-src 'none'; connect-src "
 63            "'self'; media-src 'none'; frame-src 'none'; worker-src 'none'; "
 64            "manifest-src 'none'")
 65        self.set_header("X-Content-Type-Options", "nosniff")
 66
 67    # pylint: disable=invalid-overridden-method
 68    async def prepare(self) -> None:
 69        """Prepare the handler.
 70
 71        Check if the request is authorized.
 72        """
 73        if self.request.body == b"":
 74            self.set_status(400, "No data provided.")
 75            self.write({"error": "No data provided."})
 76            self.finish()
 77            return
 78        if self.request.headers.get("ticketsplus-api-auth") is None:
 79            self.set_status(401, "No authentication token provided.")
 80            self.write({"error": "No authentication token provided."})
 81            self.finish()
 82            return
 83        if self.request.headers.get("ticketsplus-api-auth") != self._bt.stat_confg.getitem("auth_token"):
 84            self.set_status(401, "Invalid authentication token.")
 85            self.write({"error": "Invalid authentication token."})
 86            self.finish()
 87            return
 88        if self.request.headers.get("Content-Type") is None:
 89            self.set_status(400, "No Content-Type header provided.")
 90            self.write({"error": "No Content-Type header provided."})
 91            self.finish()
 92            return
 93        if self.request.headers.get("Content-Type") == "application/json":
 94            self.args = json.loads(self.request.body.decode("utf-8"))
 95            await self._bt.wait_until_ready()
 96            return
 97        self.set_status(400, "Invalid Content-Type header provided.")
 98        self.write({"error": "Invalid Content-Type header provided."})
 99        self.finish()
100
101
102class TicketHandler(BotHandler):
103    """Handles integration-based ticket creation.
104
105    The integration-based version of
106    `tickets_plus.cogs.events.on_channel_create`.
107    Does the same thing, but also parses the data from the
108    request.
109    """
110
111    async def post(self) -> None:
112        """Handle the request.
113
114        Handle the request and create the ticket.
115        Parses the POST data.
116
117        Args:
118            guild_id (str): The discord ID of the guild.
119            user_id (str): The user ID of the user.
120            ticket_channel_id (str): The ID of the channel
121        """
122        async with self._bt.get_connection() as db:
123            try:
124                guild_id = int(self.args["guild_id"])
125                user_id = int(self.args["user_id"])
126                ticket_channel_id = int(self.args["ticket_channel_id"])
127                is_new_ticket = bool(self.args["is_new_ticket"])
128            except (ValueError, KeyError):
129                self.set_status(400, "Missing or invalid parameters.")
130                self.write({"error": "Missing or invalid parameters."})
131                self.finish()
132                return
133            if not is_new_ticket:
134                self.set_status(202, "Not a new ticket.")
135                self.write({"notice": "Not a new ticket."})
136                self.finish()
137                return
138            guild = self._bt.get_guild(guild_id)
139            if guild is None:
140                self.set_status(404, "Guild not found.")
141                self.write({"error": "Guild not found."})
142                self.finish()
143                return
144            gld = await db.get_guild(
145                guild_id,
146                (
147                    orm.selectinload(models.Guild.observers_roles),
148                    orm.selectinload(models.Guild.community_roles),
149                    orm.selectinload(models.Guild.community_pings),
150                ),
151            )
152            if not gld.integrated:
153                self.set_status(409, "Guild not integrated.")
154                self.write({"error": "Guild not integrated."})
155                self.finish()
156                return
157            channel = guild.get_channel(ticket_channel_id)
158            if channel is None or not isinstance(channel, discord.TextChannel):
159                self.set_status(404, "Channel not found.")
160                self.write({"error": "Channel not found."})
161                self.finish()
162                return
163            user = self._bt.get_user(user_id)
164            self.set_status(200, "OK")
165            self.finish()
166            await events.Events.ticket_creation(self, db, (guild, gld), channel, user)
167
168
169class OverrideHandler(BotHandler):
170    """Basic messaging capabilities with the bot
171
172    Handles override messages being sent.
173    """
174
175    async def post(self):
176        """Handles override attempts.
177
178        Tries to meet the parameters specified in the attempt.
179
180        Args:
181            guild_id (str): The ID of the guild.
182            channel_id (str): The ID of the channel.
183            message (str): The message to send.
184        """
185        try:
186            guild_id = int(self.args["guild_id"])
187            channel_id = int(self.args["channel_id"])
188            message = self.args["message"]
189        except (ValueError, KeyError):
190            self.set_status(400, "Missing or invalid parameters.")
191            self.finish()
192            return
193        guild = self._bt.get_guild(guild_id)
194        if guild is None:
195            self.set_status(404, "Guild not found.")
196            self.finish()
197            return
198        channel = guild.get_channel(channel_id)
199        if channel is None or not isinstance(channel, discord.TextChannel):
200            self.set_status(404, "Channel not found.")
201            self.finish()
202            return
203        await channel.send(message)
204        self.set_status(200, "OK")
205        self.finish()

class BotHandler(tornado.web.RequestHandler): View Source

 35class BotHandler(web.RequestHandler):
 36    """Handler for the bot to send data to the app.
 37
 38    This handler is used to receive data into the bot
 39    """
 40
 41    def initialize(self, bot_instance: bot.TicketsPlusBot) -> None:
 42        """Initialize the handler.
 43
 44        Initialize the handler with the bot object.
 45
 46        Args:
 47            bot_instance: The bot object.
 48        """
 49        self._bt = bot_instance
 50        self.SUPPORTED_METHODS = ("POST",)  # pylint: disable=invalid-name
 51
 52    def set_default_headers(self) -> None:
 53        """Sets the return type to JSON.
 54
 55        Not striclty necessary, but it's good practice.
 56        """
 57        self.set_header("Content-Type", "application/json")
 58        # pylint: disable=line-too-long # skipcq: FLK-E501
 59        self.set_header("Strict-Transport-Security", "max-age=63072000; includeSubDomains; preload")
 60        # pylint: disable=line-too-long # skipcq: FLK-E501
 61        self.set_header(
 62            "Content-Security-Policy", "default-src 'none'; script-src 'none'; object-src 'none'; "
 63            "style-src 'none'; img-src 'none'; font-src 'none'; connect-src "
 64            "'self'; media-src 'none'; frame-src 'none'; worker-src 'none'; "
 65            "manifest-src 'none'")
 66        self.set_header("X-Content-Type-Options", "nosniff")
 67
 68    # pylint: disable=invalid-overridden-method
 69    async def prepare(self) -> None:
 70        """Prepare the handler.
 71
 72        Check if the request is authorized.
 73        """
 74        if self.request.body == b"":
 75            self.set_status(400, "No data provided.")
 76            self.write({"error": "No data provided."})
 77            self.finish()
 78            return
 79        if self.request.headers.get("ticketsplus-api-auth") is None:
 80            self.set_status(401, "No authentication token provided.")
 81            self.write({"error": "No authentication token provided."})
 82            self.finish()
 83            return
 84        if self.request.headers.get("ticketsplus-api-auth") != self._bt.stat_confg.getitem("auth_token"):
 85            self.set_status(401, "Invalid authentication token.")
 86            self.write({"error": "Invalid authentication token."})
 87            self.finish()
 88            return
 89        if self.request.headers.get("Content-Type") is None:
 90            self.set_status(400, "No Content-Type header provided.")
 91            self.write({"error": "No Content-Type header provided."})
 92            self.finish()
 93            return
 94        if self.request.headers.get("Content-Type") == "application/json":
 95            self.args = json.loads(self.request.body.decode("utf-8"))
 96            await self._bt.wait_until_ready()
 97            return
 98        self.set_status(400, "Invalid Content-Type header provided.")
 99        self.write({"error": "Invalid Content-Type header provided."})
100        self.finish()

Handler for the bot to send data to the app.

This handler is used to receive data into the bot

def initialize(self, bot_instance: tickets_plus.bot.TicketsPlusBot) -> None: View Source

41    def initialize(self, bot_instance: bot.TicketsPlusBot) -> None:
42        """Initialize the handler.
43
44        Initialize the handler with the bot object.
45
46        Args:
47            bot_instance: The bot object.
48        """
49        self._bt = bot_instance
50        self.SUPPORTED_METHODS = ("POST",)  # pylint: disable=invalid-name

Hook for subclass initialization. Called for each request.

A dictionary passed as the third argument of a URLSpec will be supplied as keyword arguments to initialize().

Example::

class ProfileHandler(RequestHandler):
    def initialize(self, database):
        self.database = database

    def get(self, username):
        ...

app = Application([
    (r'/user/(.*)', ProfileHandler, dict(database=database)),
    ])

def set_default_headers(self) -> None: View Source

52    def set_default_headers(self) -> None:
53        """Sets the return type to JSON.
54
55        Not striclty necessary, but it's good practice.
56        """
57        self.set_header("Content-Type", "application/json")
58        # pylint: disable=line-too-long # skipcq: FLK-E501
59        self.set_header("Strict-Transport-Security", "max-age=63072000; includeSubDomains; preload")
60        # pylint: disable=line-too-long # skipcq: FLK-E501
61        self.set_header(
62            "Content-Security-Policy", "default-src 'none'; script-src 'none'; object-src 'none'; "
63            "style-src 'none'; img-src 'none'; font-src 'none'; connect-src "
64            "'self'; media-src 'none'; frame-src 'none'; worker-src 'none'; "
65            "manifest-src 'none'")
66        self.set_header("X-Content-Type-Options", "nosniff")

Sets the return type to JSON.

Not striclty necessary, but it's good practice.

async def prepare(self) -> None: View Source

 69    async def prepare(self) -> None:
 70        """Prepare the handler.
 71
 72        Check if the request is authorized.
 73        """
 74        if self.request.body == b"":
 75            self.set_status(400, "No data provided.")
 76            self.write({"error": "No data provided."})
 77            self.finish()
 78            return
 79        if self.request.headers.get("ticketsplus-api-auth") is None:
 80            self.set_status(401, "No authentication token provided.")
 81            self.write({"error": "No authentication token provided."})
 82            self.finish()
 83            return
 84        if self.request.headers.get("ticketsplus-api-auth") != self._bt.stat_confg.getitem("auth_token"):
 85            self.set_status(401, "Invalid authentication token.")
 86            self.write({"error": "Invalid authentication token."})
 87            self.finish()
 88            return
 89        if self.request.headers.get("Content-Type") is None:
 90            self.set_status(400, "No Content-Type header provided.")
 91            self.write({"error": "No Content-Type header provided."})
 92            self.finish()
 93            return
 94        if self.request.headers.get("Content-Type") == "application/json":
 95            self.args = json.loads(self.request.body.decode("utf-8"))
 96            await self._bt.wait_until_ready()
 97            return
 98        self.set_status(400, "Invalid Content-Type header provided.")
 99        self.write({"error": "Invalid Content-Type header provided."})
100        self.finish()

Prepare the handler.

Check if the request is authorized.

class TicketHandler(BotHandler): View Source

103class TicketHandler(BotHandler):
104    """Handles integration-based ticket creation.
105
106    The integration-based version of
107    `tickets_plus.cogs.events.on_channel_create`.
108    Does the same thing, but also parses the data from the
109    request.
110    """
111
112    async def post(self) -> None:
113        """Handle the request.
114
115        Handle the request and create the ticket.
116        Parses the POST data.
117
118        Args:
119            guild_id (str): The discord ID of the guild.
120            user_id (str): The user ID of the user.
121            ticket_channel_id (str): The ID of the channel
122        """
123        async with self._bt.get_connection() as db:
124            try:
125                guild_id = int(self.args["guild_id"])
126                user_id = int(self.args["user_id"])
127                ticket_channel_id = int(self.args["ticket_channel_id"])
128                is_new_ticket = bool(self.args["is_new_ticket"])
129            except (ValueError, KeyError):
130                self.set_status(400, "Missing or invalid parameters.")
131                self.write({"error": "Missing or invalid parameters."})
132                self.finish()
133                return
134            if not is_new_ticket:
135                self.set_status(202, "Not a new ticket.")
136                self.write({"notice": "Not a new ticket."})
137                self.finish()
138                return
139            guild = self._bt.get_guild(guild_id)
140            if guild is None:
141                self.set_status(404, "Guild not found.")
142                self.write({"error": "Guild not found."})
143                self.finish()
144                return
145            gld = await db.get_guild(
146                guild_id,
147                (
148                    orm.selectinload(models.Guild.observers_roles),
149                    orm.selectinload(models.Guild.community_roles),
150                    orm.selectinload(models.Guild.community_pings),
151                ),
152            )
153            if not gld.integrated:
154                self.set_status(409, "Guild not integrated.")
155                self.write({"error": "Guild not integrated."})
156                self.finish()
157                return
158            channel = guild.get_channel(ticket_channel_id)
159            if channel is None or not isinstance(channel, discord.TextChannel):
160                self.set_status(404, "Channel not found.")
161                self.write({"error": "Channel not found."})
162                self.finish()
163                return
164            user = self._bt.get_user(user_id)
165            self.set_status(200, "OK")
166            self.finish()
167            await events.Events.ticket_creation(self, db, (guild, gld), channel, user)

Handles integration-based ticket creation.

The integration-based version of tickets_plus.cogs.events.on_channel_create. Does the same thing, but also parses the data from the request.

async def post(self) -> None: View Source

112    async def post(self) -> None:
113        """Handle the request.
114
115        Handle the request and create the ticket.
116        Parses the POST data.
117
118        Args:
119            guild_id (str): The discord ID of the guild.
120            user_id (str): The user ID of the user.
121            ticket_channel_id (str): The ID of the channel
122        """
123        async with self._bt.get_connection() as db:
124            try:
125                guild_id = int(self.args["guild_id"])
126                user_id = int(self.args["user_id"])
127                ticket_channel_id = int(self.args["ticket_channel_id"])
128                is_new_ticket = bool(self.args["is_new_ticket"])
129            except (ValueError, KeyError):
130                self.set_status(400, "Missing or invalid parameters.")
131                self.write({"error": "Missing or invalid parameters."})
132                self.finish()
133                return
134            if not is_new_ticket:
135                self.set_status(202, "Not a new ticket.")
136                self.write({"notice": "Not a new ticket."})
137                self.finish()
138                return
139            guild = self._bt.get_guild(guild_id)
140            if guild is None:
141                self.set_status(404, "Guild not found.")
142                self.write({"error": "Guild not found."})
143                self.finish()
144                return
145            gld = await db.get_guild(
146                guild_id,
147                (
148                    orm.selectinload(models.Guild.observers_roles),
149                    orm.selectinload(models.Guild.community_roles),
150                    orm.selectinload(models.Guild.community_pings),
151                ),
152            )
153            if not gld.integrated:
154                self.set_status(409, "Guild not integrated.")
155                self.write({"error": "Guild not integrated."})
156                self.finish()
157                return
158            channel = guild.get_channel(ticket_channel_id)
159            if channel is None or not isinstance(channel, discord.TextChannel):
160                self.set_status(404, "Channel not found.")
161                self.write({"error": "Channel not found."})
162                self.finish()
163                return
164            user = self._bt.get_user(user_id)
165            self.set_status(200, "OK")
166            self.finish()
167            await events.Events.ticket_creation(self, db, (guild, gld), channel, user)

Handle the request.

Handle the request and create the ticket. Parses the POST data.

Arguments:

guild_id (str): The discord ID of the guild.
user_id (str): The user ID of the user.
ticket_channel_id (str): The ID of the channel

Inherited Members

BotHandler: initialize; set_default_headers; prepare

class OverrideHandler(BotHandler): View Source

170class OverrideHandler(BotHandler):
171    """Basic messaging capabilities with the bot
172
173    Handles override messages being sent.
174    """
175
176    async def post(self):
177        """Handles override attempts.
178
179        Tries to meet the parameters specified in the attempt.
180
181        Args:
182            guild_id (str): The ID of the guild.
183            channel_id (str): The ID of the channel.
184            message (str): The message to send.
185        """
186        try:
187            guild_id = int(self.args["guild_id"])
188            channel_id = int(self.args["channel_id"])
189            message = self.args["message"]
190        except (ValueError, KeyError):
191            self.set_status(400, "Missing or invalid parameters.")
192            self.finish()
193            return
194        guild = self._bt.get_guild(guild_id)
195        if guild is None:
196            self.set_status(404, "Guild not found.")
197            self.finish()
198            return
199        channel = guild.get_channel(channel_id)
200        if channel is None or not isinstance(channel, discord.TextChannel):
201            self.set_status(404, "Channel not found.")
202            self.finish()
203            return
204        await channel.send(message)
205        self.set_status(200, "OK")
206        self.finish()

Basic messaging capabilities with the bot

Handles override messages being sent.

async def post(self): View Source

176    async def post(self):
177        """Handles override attempts.
178
179        Tries to meet the parameters specified in the attempt.
180
181        Args:
182            guild_id (str): The ID of the guild.
183            channel_id (str): The ID of the channel.
184            message (str): The message to send.
185        """
186        try:
187            guild_id = int(self.args["guild_id"])
188            channel_id = int(self.args["channel_id"])
189            message = self.args["message"]
190        except (ValueError, KeyError):
191            self.set_status(400, "Missing or invalid parameters.")
192            self.finish()
193            return
194        guild = self._bt.get_guild(guild_id)
195        if guild is None:
196            self.set_status(404, "Guild not found.")
197            self.finish()
198            return
199        channel = guild.get_channel(channel_id)
200        if channel is None or not isinstance(channel, discord.TextChannel):
201            self.set_status(404, "Channel not found.")
202            self.finish()
203            return
204        await channel.send(message)
205        self.set_status(200, "OK")
206        self.finish()

Handles override attempts.

Tries to meet the parameters specified in the attempt.

Arguments:

guild_id (str): The ID of the guild.
channel_id (str): The ID of the channel.
message (str): The message to send.

Inherited Members

BotHandler: initialize; set_default_headers; prepare