Compare commits

..

2 Commits

Author SHA1 Message Date
Tim Van Baak 5e051e7e89 Add check for duplicate memberships 2021-05-31 12:13:37 -07:00
Tim Van Baak a21092b7e0 Touch up db submodule 2021-05-31 12:13:23 -07:00
5 changed files with 62 additions and 9 deletions

View File

@ -2,6 +2,8 @@
Membership query interface Membership query interface
""" """
from sqlalchemy import select, func
from amanuensis.db import DbContext, Membership from amanuensis.db import DbContext, Membership
from amanuensis.errors import ArgumentError from amanuensis.errors import ArgumentError
@ -14,7 +16,7 @@ def create(
""" """
Create a new user membership in a lexicon. Create a new user membership in a lexicon.
""" """
# Quick argument verification # Verify argument types are correct
if not isinstance(user_id, int): if not isinstance(user_id, int):
raise ArgumentError('user_id') raise ArgumentError('user_id')
if not isinstance(lexicon_id, int): if not isinstance(lexicon_id, int):
@ -22,6 +24,14 @@ def create(
if not isinstance(is_editor, bool): if not isinstance(is_editor, bool):
raise ArgumentError('is_editor') raise ArgumentError('is_editor')
# Verify user has not already joined lexicon
if db(
select(func.count(Membership.id))
.where(Membership.user_id == user_id)
.where(Membership.lexicon_id == lexicon_id)
).scalar() > 0:
raise ArgumentError('User is already a member of lexicon')
new_membership = Membership( new_membership = Membership(
user_id=user_id, user_id=user_id,
lexicon_id=lexicon_id, lexicon_id=lexicon_id,

View File

@ -13,3 +13,19 @@ from .models import (
ArticleContentRule, ArticleContentRule,
Post, Post,
) )
__all__ = [
'DbContext',
'User',
'Lexicon',
'Membership',
'Character',
'ArticleState',
'Article',
'IndexType',
'ArticleIndex',
'ArticleIndexRule',
'ArticleContentRuleType',
'ArticleContentRule',
'Post',
]

View File

@ -33,5 +33,10 @@ class DbContext():
# Create a thread-safe session factory # Create a thread-safe session factory
self.session = scoped_session(sessionmaker(bind=self.engine)) self.session = scoped_session(sessionmaker(bind=self.engine))
def __call__(self, *args, **kwargs):
"""Provides shortcut access to session.execute."""
return self.session.execute(*args, **kwargs)
def create_all(self): def create_all(self):
"""Initializes the database schema."""
ModelBase.metadata.create_all(self.engine) ModelBase.metadata.create_all(self.engine)

View File

@ -2,6 +2,8 @@
Data model SQL definitions Data model SQL definitions
""" """
import enum import enum
import uuid
from sqlalchemy import ( from sqlalchemy import (
Boolean, Boolean,
Column, Column,
@ -17,7 +19,7 @@ from sqlalchemy import (
TypeDecorator, TypeDecorator,
) )
from sqlalchemy.orm import relationship, backref from sqlalchemy.orm import relationship, backref
import uuid from sqlalchemy.sql.schema import UniqueConstraint
from .database import ModelBase from .database import ModelBase
@ -233,6 +235,9 @@ class Membership(ModelBase):
Represents a user's participation in a Lexicon game. Represents a user's participation in a Lexicon game.
""" """
__tablename__ = 'membership' __tablename__ = 'membership'
__table_args__ = (
UniqueConstraint('user_id', 'lexicon_id'),
)
################### ###################
# Membership keys # # Membership keys #

View File

@ -1,4 +1,9 @@
from amanuensis.db import DbContext import pytest
from sqlalchemy import select
from amanuensis.db import *
from amanuensis.errors import ArgumentError
import amanuensis.backend.membership as memq import amanuensis.backend.membership as memq
@ -15,10 +20,22 @@ def test_create_membership(db: DbContext, make_user, make_lexicon):
assert mem, 'Failed to create membership' assert mem, 'Failed to create membership'
# Check that the user and lexicon are mutually visible in the ORM relationships # Check that the user and lexicon are mutually visible in the ORM relationships
assert new_user.memberships, 'User memberships not updated' assert any(map(lambda mem: mem.lexicon == new_lexicon, new_user.memberships))
assert new_lexicon.memberships, 'Lexicon memberships not updated' assert any(map(lambda mem: mem.user == new_user, new_lexicon.memberships))
assert new_user.memberships[0].lexicon_id == new_lexicon.id
assert new_lexicon.memberships[0].user_id == new_user.id
# Check that the editor flag was set properly # Check that the editor flag was set properly
assert new_lexicon.memberships editor = db(
select(User)
.join(User.memberships)
.join(Membership.lexicon)
.where(Lexicon.id == new_lexicon.id)
.where(Membership.is_editor == True)
).scalar_one()
assert editor is not None
assert isinstance(editor, User)
assert editor.id == new_user.id
# Check that joining twice is not allowed
with pytest.raises(ArgumentError):
mem2 = memq.create(db, new_user.id, new_lexicon.id, False)
assert mem2