-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathdb.py
62 lines (49 loc) · 1.81 KB
/
db.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
"""Here are database table classes defined with SQLAlchemy ORM.
"""
from sqlalchemy import Column
from sqlalchemy import DateTime
from sqlalchemy import Index
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
PROVIDERS: "name: prefix" = {
"Git": "git",
"Gist": "gist",
"GitHub": "gh",
"GitLab": "gl",
"Zenodo": "zenodo",
"Figshare": "figshare",
"Hydroshare": "hydroshare",
"Dataverse": "dataverse",
}
Base = declarative_base()
class Launch(Base):
"""Table for MyBinder launch events
- https://archive.analytics.mybinder.org/
- https://binderhub.readthedocs.io/en/latest/eventlogging.html
"""
__tablename__ = "launches"
# this is a fake primary key for sqlalchemy only, it doesnt exist in db
id = Column(Integer, primary_key=True)
timestamp = Column(DateTime(timezone=True), nullable=False)
provider = Column(String, nullable=False)
spec = Column(String, nullable=False)
repo = Column(String, nullable=False)
ref = Column(String, nullable=False) # unresolved ref
resolved_ref = Column(String, nullable=False) # ref in archives
origin = Column(String, nullable=False)
schema = Column(String, nullable=False)
version = Column(Integer, nullable=False)
status = Column(String, nullable=False)
__table_args__ = (
Index("launches_timestamp_idx", timestamp.desc()),
Index("launches_provider_idx", provider),
Index("launches_repo_idx", repo),
Index("launches_origin_idx", origin),
Index("launches_provider_repo_idx", provider, repo),
)
def __repr__(self):
return f"<Launch({self.provider=}, {self.repo=}, {self.timestamp=})>"
@property
def provider_prefix(self) -> str:
return PROVIDERS[self.provider]