322 lines
9.4 KiB
Python
322 lines
9.4 KiB
Python
"""Database models."""
|
|
|
|
import os
|
|
import typing
|
|
|
|
import sqlalchemy
|
|
import sqlalchemy.orm.decl_api
|
|
from sqlalchemy import Index, func, text
|
|
from sqlalchemy.dialects import postgresql
|
|
from sqlalchemy.dialects.postgresql import TSVECTOR
|
|
from sqlalchemy.ext.associationproxy import association_proxy
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.ext.orderinglist import ordering_list
|
|
from sqlalchemy.orm import relationship
|
|
from sqlalchemy.schema import Column, ForeignKey
|
|
from sqlalchemy.types import Boolean, Date, DateTime, Integer, String
|
|
|
|
from .database import session
|
|
|
|
Base: sqlalchemy.orm.decl_api.DeclarativeMeta = declarative_base()
|
|
Base.query = session.query_property()
|
|
|
|
|
|
class TimeStampedModel(Base):
|
|
"""Time stamped model."""
|
|
|
|
__abstract__ = True
|
|
created = Column(DateTime, default=func.now())
|
|
modified = Column(DateTime, default=func.now(), onupdate=func.now())
|
|
|
|
|
|
class Series(TimeStampedModel):
|
|
"""Conference series."""
|
|
|
|
__tablename__ = "series"
|
|
id = Column(Integer, primary_key=True)
|
|
name = Column(String, nullable=False)
|
|
slug = Column(String, unique=True)
|
|
wikidata_qid = Column(String, unique=True)
|
|
|
|
conferences = relationship("Conference", back_populates="series")
|
|
|
|
|
|
class Conference(TimeStampedModel):
|
|
"""Conference."""
|
|
|
|
__tablename__ = "conference"
|
|
id = Column(Integer, primary_key=True)
|
|
title = Column(String, nullable=False)
|
|
start = Column(Date)
|
|
end = Column(Date)
|
|
days = Column(Integer)
|
|
timezone = Column(String)
|
|
location = Column(String)
|
|
country = Column(String)
|
|
acronym = Column(String)
|
|
url = Column(String)
|
|
schedule_xml_url = Column(String)
|
|
short_name = Column(String, unique=True)
|
|
venue_id = Column(Integer, ForeignKey("venue.id"))
|
|
online = Column(Boolean)
|
|
wikidata_qid = Column(String, unique=True)
|
|
series_id = Column(Integer, ForeignKey("series.id"))
|
|
|
|
people_detail = relationship(
|
|
"ConferencePerson", lazy="dynamic", back_populates="conference"
|
|
)
|
|
people = association_proxy("people_detail", "person")
|
|
|
|
events = relationship(
|
|
"Event",
|
|
order_by="Event.event_date",
|
|
back_populates="conference",
|
|
lazy="dynamic",
|
|
)
|
|
|
|
venue = relationship("Venue", back_populates="conferences")
|
|
series = relationship("Series", back_populates="conferences")
|
|
|
|
|
|
class City(TimeStampedModel):
|
|
"""City."""
|
|
|
|
__tablename__ = "city"
|
|
id = Column(Integer, primary_key=True)
|
|
slug = Column(String, nullable=False, unique=True)
|
|
name = Column(String, nullable=False)
|
|
country_code = Column(String, ForeignKey("country.alpha2"), nullable=False)
|
|
wikidata_qid = Column(String, nullable=False)
|
|
|
|
venues = relationship("Venue", back_populates="city")
|
|
country = relationship("Country", back_populates="cities")
|
|
|
|
|
|
class Venue(TimeStampedModel):
|
|
"""Venue."""
|
|
|
|
__tablename__ = "venue"
|
|
id = Column(Integer, primary_key=True)
|
|
name = Column(String, nullable=False)
|
|
city_id = Column(Integer, ForeignKey("city.id"))
|
|
wikidata_qid = Column(String, nullable=False)
|
|
|
|
conferences = relationship("Conference", back_populates="venue")
|
|
city = relationship("City", back_populates="venues")
|
|
|
|
|
|
class Country(TimeStampedModel):
|
|
"""Country."""
|
|
|
|
__tablename__ = "country"
|
|
alpha2 = Column(String, primary_key=True)
|
|
name = Column(String)
|
|
wikidata_qid = Column(String, nullable=False)
|
|
|
|
cities = relationship("City", back_populates="country")
|
|
|
|
@property
|
|
def flag(self) -> str:
|
|
a = ord("A")
|
|
flag_a = 0x1F1E6
|
|
char1, char2 = (
|
|
flag_a + ord(self.alpha2[0]) - a,
|
|
flag_a + ord(self.alpha2[1]) - a,
|
|
)
|
|
|
|
return chr(char1) + chr(char2)
|
|
|
|
|
|
class ConferencePerson(Base):
|
|
__tablename__ = "conference_person"
|
|
conference_id = Column(Integer, ForeignKey("conference.id"), primary_key=True)
|
|
person_id = Column(Integer, ForeignKey("person.id"), primary_key=True)
|
|
named_as = Column(String)
|
|
bio = Column(String)
|
|
slug = Column(String)
|
|
url = Column(String)
|
|
affiliation = Column(String)
|
|
photo_url = Column(String)
|
|
|
|
person = relationship("Person", back_populates="conferences_association")
|
|
conference = relationship("Conference", back_populates="people_detail")
|
|
|
|
@property
|
|
def events(self):
|
|
return (
|
|
Event.query.join(EventPerson)
|
|
.filter(
|
|
Event.conference == self.conference, EventPerson.person == self.person
|
|
)
|
|
.order_by(Event.event_date.desc())
|
|
)
|
|
|
|
|
|
class Event(TimeStampedModel):
|
|
"""Event."""
|
|
|
|
__tablename__ = "event"
|
|
id = Column(Integer, primary_key=True)
|
|
conference_id = Column(Integer, ForeignKey("conference.id"), nullable=False)
|
|
event_date = Column(DateTime)
|
|
# day = Column(Integer)
|
|
guid = Column(String)
|
|
start = Column(String)
|
|
duration = Column(String)
|
|
room = Column(String)
|
|
track = Column(String)
|
|
slug = Column(String)
|
|
title = Column(String, nullable=False)
|
|
abstract = Column(String)
|
|
description = Column(String)
|
|
event_type = Column(String)
|
|
url = Column(String)
|
|
cancelled = Column(Boolean)
|
|
search_vector = Column(TSVECTOR)
|
|
|
|
trigger = text(
|
|
"""
|
|
CREATE TRIGGER event_vector_update BEFORE INSERT OR UPDATE
|
|
ON event FOR EACH ROW EXECUTE FUNCTION
|
|
tsvector_update_trigger(search_vector, 'pg_catalog.english', title, abstract, description);
|
|
"""
|
|
)
|
|
|
|
Index(
|
|
"event_search_vector_idx",
|
|
search_vector,
|
|
postgresql_using="gin",
|
|
postgresql_ops={"abstract_search_vector": "gin_trgm_ops"},
|
|
)
|
|
|
|
conference = relationship("Conference", back_populates="events")
|
|
|
|
people_detail = relationship(
|
|
"EventPerson",
|
|
order_by="EventPerson.position",
|
|
lazy="dynamic",
|
|
back_populates="event",
|
|
collection_class=ordering_list("position"),
|
|
)
|
|
people = association_proxy(
|
|
"people_detail",
|
|
"person",
|
|
creator=lambda i: EventPerson(person=i),
|
|
)
|
|
|
|
|
|
class Person(TimeStampedModel):
|
|
"""Person."""
|
|
|
|
__tablename__ = "person"
|
|
id = Column(Integer, primary_key=True)
|
|
name = Column(String)
|
|
wikidata_qid = Column(String)
|
|
gender = Column(String)
|
|
wikidata_photo = Column(postgresql.ARRAY(String))
|
|
|
|
events_association = relationship(
|
|
"EventPerson",
|
|
back_populates="person",
|
|
lazy="dynamic",
|
|
)
|
|
events = association_proxy("events_association", "event")
|
|
|
|
conferences_association = relationship(
|
|
"ConferencePerson", lazy="dynamic", back_populates="person"
|
|
)
|
|
conferences = association_proxy("conferences_association", "conference")
|
|
|
|
@property
|
|
def conference_count(self):
|
|
return ConferencePerson.query.filter_by(person_id=self.id).count()
|
|
|
|
@property
|
|
def event_count(self):
|
|
return EventPerson.query.filter_by(person_id=self.id).count()
|
|
|
|
def active_years(self):
|
|
q = (
|
|
session.query(func.min(Event.event_date), func.max(Event.event_date))
|
|
.join(EventPerson)
|
|
.filter_by(person_id=self.id)
|
|
)
|
|
return q.one()
|
|
|
|
# photos = relationship("PersonPhoto", back_populates="person")
|
|
|
|
def events_by_time(self):
|
|
q = (
|
|
session.query(Event)
|
|
.join(EventPerson)
|
|
.filter(EventPerson.person == self)
|
|
.order_by(Event.event_date.desc())
|
|
)
|
|
|
|
return q
|
|
|
|
def conference_by_time(self):
|
|
q = (
|
|
session.query(ConferencePerson)
|
|
.join(Conference)
|
|
.filter(ConferencePerson.person == self)
|
|
.order_by(Conference.start.desc())
|
|
)
|
|
|
|
return q
|
|
|
|
def bio_source(self) -> ConferencePerson | None:
|
|
bio_list = [cp for cp in self.conferences_association if cp.bio]
|
|
|
|
if not bio_list:
|
|
return None
|
|
|
|
if len(bio_list) == 1:
|
|
return typing.cast(ConferencePerson, bio_list[0])
|
|
|
|
recent = max(bio_list, key=lambda cp: cp.conference.start)
|
|
|
|
len_recent_bio = len(recent.bio)
|
|
longest = max(bio_list, key=lambda cp: len(cp.bio))
|
|
|
|
if recent == longest:
|
|
return typing.cast(ConferencePerson, recent)
|
|
|
|
best = longest if len(longest.bio) > len_recent_bio * 2 else recent
|
|
return typing.cast(ConferencePerson, best)
|
|
|
|
def photo_filename(self) -> str | None:
|
|
if self.wikidata_photo:
|
|
return os.path.join("wikidata_photo", "thumb", self.wikidata_photo[0])
|
|
|
|
q = self.conferences_association.filter(ConferencePerson.photo_url.isnot(None))
|
|
if q.count() == 0:
|
|
return None
|
|
|
|
best = max(q, key=lambda cp: cp.conference.start)
|
|
ext = best.photo_url.rpartition(".")[-1]
|
|
filename = f"{best.conference_id}_{self.id}.{ext}"
|
|
return os.path.join("conference_photo", filename)
|
|
|
|
|
|
# class PersonPhoto(TimeStampedModel):
|
|
# """Person photo."""
|
|
#
|
|
# __tablename__ = "person_photo"
|
|
# person_id = Column(Integer, ForeignKey("person.id"), primary_key=True)
|
|
# source_filename = Column(String)
|
|
#
|
|
# person = relationship("Person", back_populates="photos")
|
|
|
|
|
|
class EventPerson(Base):
|
|
"""Event person."""
|
|
|
|
__tablename__ = "event_person"
|
|
event_id = Column(Integer, ForeignKey("event.id"), primary_key=True)
|
|
person_id = Column(Integer, ForeignKey("person.id"), primary_key=True)
|
|
position = Column(Integer, nullable=False)
|
|
|
|
person = relationship("Person", back_populates="events_association")
|
|
event = relationship("Event", back_populates="people_detail")
|