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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
import os
import sys
from bson.objectid import ObjectId
from datetime import datetime
import sqlalchemy as db
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
class Types(object):
class ObjectId(db.TypeDecorator):
impl = db.UnicodeText
def process_bind_param(self, value : ObjectId, dialect) -> str:
return str(value)
def process_result_value(self, value : str, dialect) -> ObjectId:
if value in [None, "None"]:
return ObjectId("0" * 24)
return ObjectId(value)
class List(db.TypeDecorator):
impl = db.UnicodeText
def process_bind_param(self, value : list, dialect) -> str:
return ";".join(i.replace(";", ",") for i in value)
def process_result_value(self, value : str, dialect) -> list:
return value.split(";")
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = db.Column(Types.ObjectId, primary_key=True, default=ObjectId)
name = db.Column(db.UnicodeText, nullable=False)
email = db.Column(db.UnicodeText)
hash = db.Column(db.String(512), nullable=False)
class Upload(Base):
__tablename__ = "uploads"
id = db.Column(Types.ObjectId, primary_key=True, default=ObjectId)
title = db.Column(db.UnicodeText, nullable=False)
author_id = db.Column(Types.ObjectId, db.ForeignKey("users.id"), default=lambda: ObjectId("0"*24))
description = db.Column(db.UnicodeText, default="")
slug = db.Column(db.Text, nullable=False)
tags = db.Column(Types.List, default=list)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
updated_at = db.Column(db.DateTime, nullable=False, default=datetime.now, onupdate=datetime.now)
_v = db.Column(db.Integer, default=0)
author = relationship("User")
#dependency = relationship("Dependency")
class File(Base):
__tablename__ = "files"
id = db.Column(Types.ObjectId, primary_key=True, default=ObjectId)
name = db.Column(db.Unicode(255), nullable=False)
hash = db.Column(db.String(40), nullable=False)
content_type = db.Column(db.String(255), default="application/octet-stream")
length = db.Column(db.BigInteger, default=0)
date = db.Column(db.DateTime, nullable=False, default=datetime.now)
upload_id = db.Column(Types.ObjectId, db.ForeignKey("uploads.id"))
download_url = db.Column(db.Text)
upload = relationship("Upload")
class Comment(Base):
__tablename__ = "comments"
id = db.Column(Types.ObjectId, primary_key=True, default=ObjectId)
author_id = db.Column(Types.ObjectId, db.ForeignKey("users.id"))
upload_id = db.Column(Types.ObjectId, db.ForeignKey("uploads.id"))
created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
updated_at = db.Column(db.DateTime, nullable=False, default=datetime.now, onupdate=datetime.now)
body = db.Column(db.UnicodeText, nullable=False)
author = relationship("User")
upload = relationship("Upload")
class Voting(Base):
__tablename__ = "votings"
id = db.Column(Types.ObjectId, primary_key=True, default=ObjectId)
author_id = db.Column(Types.ObjectId, db.ForeignKey("users.id"))
target_id = db.Column(Types.ObjectId, db.ForeignKey("uploads.id"))
author = relationship("User")
target = relationship("Upload")
pass
#class Dependency(Base):
#__tablename__ = "dependencies"
#id = db.Column(db.Integer, primary_key=True, default=ObjectId)
#target_id = db.Column(Types.ObjectId, db.ForeignKey("uploads.id"))
##dependency_id = db.Column(Types.ObjectId, db.ForeignKey("uploads.id"))
#target = relationship("Upload", foreign_keys=[target_id])
##dependency = relationship("Upload", foreign_keys=[dependency_id])
engine = db.create_engine("sqlite:///parry.db")
Base.metadata.bind = engine
Base.metadata.create_all(engine)
DBSession = sessionmaker(bind=engine)
|
