Format with black

Author: ChrisLovering

api/__init__.py

@@ -6,9 +6,7 @@
 date_format_string = "%Y-%m-%d %H:%M:%S %z"
 
 logging.basicConfig(
- format=format_string,
- datefmt=date_format_string,
- level=getattr(logging, constants.Config.LOG_LEVEL.upper())
+ format=format_string, datefmt=date_format_string, level=getattr(logging, constants.Config.LOG_LEVEL.upper())
 )
 
 logging.getLogger().info("Logging initialization complete")

api/database.py

@@ -15,9 +15,7 @@ class TeamUser(Base):
  """A user who belongs to a team."""
 
  __tablename__ = "team_has_user"
- __table_args__ = (
- PrimaryKeyConstraint('team_id', 'user_id'),
- )
+ __table_args__ = (PrimaryKeyConstraint("team_id", "user_id"),)
 
  team_id = Column(ForeignKey("teams.id"), nullable=False)
  user_id = Column(ForeignKey("users.id"), nullable=False)
@@ -65,18 +63,14 @@ class Team(Base):
  jam = relationship("Jam", back_populates="teams", lazy="joined")
  users = relationship("TeamUser", back_populates="team", lazy="joined")
 
- __table_args__ = (
- Index('team_name_jam_unique', text("lower(name)"), "jam_id", unique=True),
- )
+ __table_args__ = (Index("team_name_jam_unique", text("lower(name)"), "jam_id", unique=True),)
 
 
 class Winner(Base):
  """A user who has won a code jam."""
 
  __tablename__ = "winners"
- __table_args__ = (
- PrimaryKeyConstraint('jam_id', 'user_id'),
- )
+ __table_args__ = (PrimaryKeyConstraint("jam_id", "user_id"),)
 
  jam_id = Column(ForeignKey("jams.id"), nullable=False)
  user_id = Column(ForeignKey("users.id"), nullable=False)

api/middleware.py

@@ -15,9 +15,7 @@ class TokenAuthentication(AuthenticationBackend):
  def __init__(self, token: str) -> None:
  self.expected_auth_header = f"Token {token}"
 
- async def authenticate(
- self, request: Request
- ) -> tuple[AuthCredentials, SimpleUser]:
+ async def authenticate(self, request: Request) -> tuple[AuthCredentials, SimpleUser]:
  """Authenticate the request based on the Authorization header."""
  if Config.DEBUG:
  credentials = AuthCredentials(scopes=["debug"])

api/routers/codejams.py

@@ -24,11 +24,7 @@ async def get_codejams(session: AsyncSession = Depends(get_db_session)) -> list[
 @router.get(
  "/{codejam_id}",
  response_model=CodeJamResponse,
- responses={
- 404: {
- "description": "CodeJam could not be found or there is no ongoing code jam."
- }
- }
+ responses={404: {"description": "CodeJam could not be found or there is no ongoing code jam."}},
 )
 async def get_codejam(codejam_id: int, session: AsyncSession = Depends(get_db_session)) -> Jam:
  """
@@ -37,9 +33,7 @@ async def get_codejam(codejam_id: int, session: AsyncSession = Depends(get_db_se
  Passing -1 as the codejam ID will return the ongoing codejam.
  """
  if codejam_id == -1:
- ongoing_jams = (
- await session.execute(select(Jam).where(Jam.ongoing == True))
- ).unique().scalars().all()
+ ongoing_jams = (await session.execute(select(Jam).where(Jam.ongoing == True))).unique().scalars().all()
 
  if not ongoing_jams:
  raise HTTPException(status_code=404, detail="There is no ongoing codejam.")
@@ -56,19 +50,12 @@ async def get_codejam(codejam_id: int, session: AsyncSession = Depends(get_db_se
  return jam
 
 
-@router.patch(
- "/{codejam_id}",
- responses={
- 404: {
- "description": "Code Jam with specified ID does not exist."
- }
- }
-)
+@router.patch("/{codejam_id}", responses={404: {"description": "Code Jam with specified ID does not exist."}})
 async def modify_codejam(
  codejam_id: int,
  name: Optional[str] = None,
  ongoing: Optional[bool] = None,
- session: AsyncSession = Depends(get_db_session)
+ session: AsyncSession = Depends(get_db_session),
 ) -> None:
  """Modify the specified codejam to change its name and/or whether it's the ongoing code jam."""
  codejam = await session.execute(select(Jam).where(Jam.id == codejam_id))
@@ -113,16 +100,19 @@ async def create_codejam(codejam: CodeJam, session: AsyncSession = Depends(get_d
  jam_id=jam.id,
  name=raw_team.name,
  discord_role_id=raw_team.discord_role_id,
- discord_channel_id=raw_team.discord_channel_id
+ discord_channel_id=raw_team.discord_channel_id,
  )
  session.add(team)
  # Flush here to receive team ID
  await session.flush()
 
  for raw_user in raw_team.users:
- if not (
- await session.execute(select(User).where(User.id == raw_user.user_id))
- ).unique().scalars().one_or_none():
+ if (
+ not (await session.execute(select(User).where(User.id == raw_user.user_id)))
+ .unique()
+ .scalars()
+ .one_or_none()
+ ):
  user = User(id=raw_user.user_id)
  session.add(user)
 

api/routers/infractions.py

@@ -22,11 +22,7 @@ async def get_infractions(session: AsyncSession = Depends(get_db_session)) -> li
 @router.get(
  "/{infraction_id}",
  response_model=InfractionResponse,
- responses={
- 404: {
- "description": "Infraction could not be found."
- }
- }
+ responses={404: {"description": "Infraction could not be found."}},
 )
 async def get_infraction(infraction_id: int, session: AsyncSession = Depends(get_db_session)) -> DbInfraction:
  """Get a specific infraction stored in the database by ID."""
@@ -40,13 +36,7 @@ async def get_infraction(infraction_id: int, session: AsyncSession = Depends(get
 
 
 @router.post(
- "/",
- response_model=InfractionResponse,
- responses={
- 404: {
- "Description": "Jam ID or User ID could not be found."
- }
- }
+ "/", response_model=InfractionResponse, responses={404: {"Description": "Jam ID or User ID could not be found."}}
 )
 async def create_infraction(infraction: Infraction, session: AsyncSession = Depends(get_db_session)) -> DbInfraction:
  """Add an infraction for a user to the database."""
@@ -61,10 +51,7 @@ async def create_infraction(infraction: Infraction, session: AsyncSession = Depe
  raise HTTPException(404, "User with specified ID could not be found.")
 
  infraction = DbInfraction(
- user_id=user_id,
- jam_id=jam_id,
- infraction_type=infraction.infraction_type,
- reason=infraction.reason
+ user_id=user_id, jam_id=jam_id, infraction_type=infraction.infraction_type, reason=infraction.reason
  )
  session.add(infraction)
  await session.flush()

api/routers/teams.py

@@ -47,15 +47,7 @@ async def get_teams(current_jam: bool = False, session: AsyncSession = Depends(g
  return teams.scalars().all()
 
 
-@router.get(
- "/find",
- response_model=TeamResponse,
- responses={
- 404: {
- "description": "Team could not be found."
- }
- }
-)
+@router.get("/find", response_model=TeamResponse, responses={404: {"description": "Team could not be found."}})
 async def find_team_by_name(
  name: str, jam_id: Optional[int] = None, session: AsyncSession = Depends(get_db_session)
 ) -> Team:
@@ -77,29 +69,13 @@ async def find_team_by_name(
  return team
 
 
-@router.get(
- "/{team_id}",
- response_model=TeamResponse,
- responses={
- 404: {
- "description": "Team could not be found."
- }
- }
-)
+@router.get("/{team_id}", response_model=TeamResponse, responses={404: {"description": "Team could not be found."}})
 async def get_team(team_id: int, session: AsyncSession = Depends(get_db_session)) -> Team:
  """Get a specific code jam team in the database by ID."""
  return await ensure_team_exists(team_id, session)
 
 
-@router.get(
- "/{team_id}/users",
- response_model=list[User],
- responses={
- 404: {
- "description": "Team could not be found."
- }
- }
-)
+@router.get("/{team_id}/users", response_model=list[User], responses={404: {"description": "Team could not be found."}})
 async def get_team_users(team_id: int, session: AsyncSession = Depends(get_db_session)) -> list[TeamUser]:
  """Get the users of a specific code jam team in the database."""
  await ensure_team_exists(team_id, session)
@@ -117,10 +93,8 @@ async def get_team_users(team_id: int, session: AsyncSession = Depends(get_db_se
  404: {
  "description": "Team or user could not be found.",
  },
- 400: {
- "description": "This user is already on the team."
- }
- }
+ 400: {"description": "This user is already on the team."},
+ },
 )
 async def add_user_to_team(
  team_id: int, user_id: int, is_leader: bool = False, session: AsyncSession = Depends(get_db_session)
@@ -151,10 +125,8 @@ async def add_user_to_team(
  404: {
  "description": "Team or user could not be found.",
  },
- 400: {
- "description": "This user is not on this team."
- }
- }
+ 400: {"description": "This user is not on this team."},
+ },
 )
 async def remove_user_from_team(
  team_id: int, user_id: int, session: AsyncSession = Depends(get_db_session)

api/routers/users.py

@@ -33,8 +33,12 @@ async def get_user_data(session: AsyncSession, user_id: int) -> dict[str, Any]:
 
  participation_history.append(
  {
- "jam_id": user_team.team.jam.id, "top_10": top_10, "first_place": first_place,
- "team_id": user_team.team.id, "is_leader": user_team.is_leader, "infractions": infractions
+ "jam_id": user_team.team.jam.id,
+ "top_10": top_10,
+ "first_place": first_place,
+ "team_id": user_team.team.id,
+ "is_leader": user_team.is_leader,
+ "infractions": infractions,
  }
  )
 
@@ -52,15 +56,7 @@ async def get_users(session: AsyncSession = Depends(get_db_session)) -> list[dic
  return [await get_user_data(session, user) for user in users.scalars().all()]
 
 
-@router.get(
- "/{user_id}",
- response_model=UserResponse,
- responses={
- 404: {
- "description": "User could not be found."
- }
- }
-)
+@router.get("/{user_id}", response_model=UserResponse, responses={404: {"description": "User could not be found."}})
 async def get_user(user_id: int, session: AsyncSession = Depends(get_db_session)) -> dict[str, Any]:
  """Get a specific user stored in the database by ID."""
  user = await session.execute(select(User).where(User.id == user_id))
@@ -72,15 +68,7 @@ async def get_user(user_id: int, session: AsyncSession = Depends(get_db_session)
  return await get_user_data(session, user_id)
 
 
-@router.post(
- "/{user_id}",
- response_model=UserResponse,
- responses={
- 400: {
- "description": "User already exists."
- }
- }
-)
+@router.post("/{user_id}", response_model=UserResponse, responses={400: {"description": "User already exists."}})
 async def create_user(user_id: int, session: AsyncSession = Depends(get_db_session)) -> dict[str, Any]:
  """Create a new user with the specified ID to the database."""
  user = await session.execute(select(User).where(User.id == user_id))
@@ -102,12 +90,10 @@ async def create_user(user_id: int, session: AsyncSession = Depends(get_db_sessi
  responses={
  404: {
  "description": (
- "User not found, "
- "there is no ongoing code jam or "
- "user isn't participating in current code jam."
+ "User not found, " "there is no ongoing code jam or " "user isn't participating in current code jam."
  )
  }
- }
+ },
 )
 async def get_current_team(user_id: int, session: AsyncSession = Depends(get_db_session)) -> dict[str, Any]:
  """Get a user's current team information."""
@@ -117,16 +103,12 @@ async def get_current_team(user_id: int, session: AsyncSession = Depends(get_db_
  if not user.scalars().one_or_none():
  raise HTTPException(status_code=404, detail="User with specified ID could not be found.")
 
- ongoing_jam = (
- await session.execute(select(Jam).where(Jam.ongoing == True))
- ).unique().scalars().one_or_none()
+ ongoing_jam = (await session.execute(select(Jam).where(Jam.ongoing == True))).unique().scalars().one_or_none()
 
  if not ongoing_jam:
  raise HTTPException(status_code=404, detail="There is no ongoing codejam.")
 
- user_teams = (
- await session.execute(select(TeamUser).where(TeamUser.user_id == user_id))
- ).unique().scalars().all()
+ user_teams = (await session.execute(select(TeamUser).where(TeamUser.user_id == user_id))).unique().scalars().all()
 
  current_team = None
  for user_team in user_teams:
@@ -135,9 +117,6 @@ async def get_current_team(user_id: int, session: AsyncSession = Depends(get_db_
  break
 
  if not current_team:
- raise HTTPException(
- status_code=404,
- detail="User with specified ID isn't participating in ongoing codejam."
- )
+ raise HTTPException(status_code=404, detail="User with specified ID isn't participating in ongoing codejam.")
 
  return current_team

api/routers/winners.py

@@ -14,11 +14,7 @@
 @router.get(
  "/{jam_id}",
  response_model=list[WinnerResponse],
- responses={
- 404: {
- "description": "The specified codejam could not be found."
- }
- }
+ responses={404: {"description": "The specified codejam could not be found."}},
 )
 async def get_winners(jam_id: int, session: AsyncSession = Depends(get_db_session)) -> list[DbWinner]:
  """Get the top ten winners from the specified codejam."""
@@ -37,16 +33,12 @@ async def get_winners(jam_id: int, session: AsyncSession = Depends(get_db_sessio
  "/{jam_id}",
  response_model=list[WinnerResponse],
  responses={
- 400: {
- "description": "The provided winners list is empty or contains duplicate users."
- },
+ 400: {"description": "The provided winners list is empty or contains duplicate users."},
  404: {
  "description": "The codejam or one of the users provided could not be found.",
  },
- 409: {
- "description": "One or more users are already a winner in the specified codejam."
- }
- }
+ 409: {"description": "One or more users are already a winner in the specified codejam."},
+ },
 )
 async def create_winners(
  jam_id: int, winners: list[Winner], session: AsyncSession = Depends(get_db_session)

migrations/env.py

@@ -50,10 +50,7 @@ def run_migrations_offline() -> None:
 def do_run_migrations(connection: AsyncConnection):
  """Run all migrations on the given connection."""
  context.configure(
- connection=connection,
- target_metadata=target_metadata,
- compare_type=True,
- compare_server_default=True
+ connection=connection, target_metadata=target_metadata, compare_type=True, compare_server_default=True
  )
 
  with context.begin_transaction():