webhooks: add impl to create and fetch webhook list

- schemas: add WEBHOOK_CREATE

litecord/blueprints/webhooks.py

@@ -17,19 +17,92 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 """
 
-from quart import Blueprint
+from typing import Dict, Any, Optional
+
+from quart import Blueprint, jsonify, current_app as app, request
+
+from litecord.auth import token_check
+from litecord.blueprints.checks import channel_check, channel_perm_check
+
+from litecord.schemas import validate, WEBHOOK_CREATE
+from litecord.enums import ChannelType
+from litecord.snowflake import get_snowflake
+from litecord.utils import async_map
 
 bp = Blueprint('webhooks', __name__)
 
 
+async def get_webhook(webhook_id: int) -> Optional[Dict[str, Any]]:
+    """Get a webhook data"""
+    row = await app.db.fetchrow("""
+    SELECT id::text, guild_id::text, channel_id::text, creator_id
+           name, avatar, token
+    FROM webhooks
+    WHERE id = $1
+    """, webhook_id)
+
+    if not row:
+        return None
+
+    drow = dict(row)
+
+    drow['user'] = await app.storage.get_user(row['creator_id'])
+    drow.pop('creator_id')
+
+    return drow
+
+
+async def _webhook_check():
+    user_id = await token_check()
+
+    await channel_check(user_id, channel_id, ChannelType.GUILD_TEXT)
+    await channel_perm_check(user_id, channel_id, 'manage_webhooks')
+
+    return user_id
+
+
 @bp.route('/channels/<int:channel_id>/webhooks', methods=['POST'])
-async def create_webhook(channel_id):
+async def create_webhook(channel_id: int):
-    pass
+    """Create a webhook given a channel."""
+    user_id = await _webhook_check()
+
+    j = validate(await request.get_json(), WEBHOOK_CREATE)
+
+    guild_id = await app.storage.guild_from_channel(channel_id)
+
+    webhook_id = get_snowflake()
+    token = 'asd'
+
+    await app.db.execute(
+        """
+        INSERT INTO webhooks
+            (id, guild_id, channel_id, creator_id, name, avatar, token)
+        VALUES
+            ($1, $2, $3, $4, $5, $6, $7)
+        """,
+        webhook_id, guild_id, channel_id, user_id,
+        j['name'], j.get('avatar'), token
+    )
+
+    return jsonify(await get_webhook(webhook_id))
 
 
 @bp.route('/channels/<int:channel_id>/webhooks', methods=['GET'])
-async def get_channel_webhook(channel_id):
+async def get_channel_webhook(channel_id: int):
-    pass
+    """Get a list of webhooks in a channel"""
+    _user_id = await _webhook_check()
+
+    webhook_ids = await app.db.fetch("""
+    SELECT id
+    FROM webhooks
+    WHERE channel_id = $1
+    """, channel_id)
+
+    webhook_ids = [r['id'] for r in webhook_ids]
+
+    return jsonify(
+        await async_map(get_webhook, webhook_ids)
+    )
 
 
 @bp.route('/guilds/<int:guild_id>/webhooks', methods=['GET'])

litecord/schemas.py

@@ -677,3 +677,11 @@ VANITY_URL_PATCH = {
     # TODO: put proper values in maybe an invite data type
     'code': {'type': 'string', 'minlength': 5, 'maxlength': 30}
 }
+
+WEBHOOK_CREATE = {
+    'name': {
+        'type': 'string', 'minlength': 2, 'maxlength': 32,
+        'required': True
+    },
+    'avatar': {'type': 'b64_icon', 'required': False, 'nullable': False}
+}