forked from GoogleCloudPlatform/getting-started-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmodel_cloudsql.py
More file actions
119 lines (90 loc) · 2.97 KB
/
Copy pathmodel_cloudsql.py
File metadata and controls
119 lines (90 loc) · 2.97 KB
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
# Copyright 2015 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy
builtin_list = list
db = SQLAlchemy()
def init_app(app):
db.init_app(app)
def from_sql(row):
"""Translates a SQLAlchemy model instance into a dictionary"""
data = row.__dict__.copy()
data['id'] = row.id
data.pop('_sa_instance_state')
return data
class Book(db.Model):
__tablename__ = 'books'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(255))
author = db.Column(db.String(255))
publishedDate = db.Column(db.String(255))
imageUrl = db.Column(db.String(255))
description = db.Column(db.String(255))
createdBy = db.Column(db.String(255))
createdById = db.Column(db.String(255))
def __repr__(self):
return "<Book(title='%s', author=%s)" % (self.title, self.author)
def list(limit=10, cursor=None):
cursor = int(cursor) if cursor else 0
query = (Book.query
.order_by(Book.title)
.limit(limit)
.offset(cursor))
books = builtin_list(map(from_sql, query.all()))
next_page = cursor + limit if len(books) == limit else None
return (books, next_page)
# [START list_by_user]
def list_by_user(user_id, limit=10, cursor=None):
cursor = int(cursor) if cursor else 0
query = (Book.query
.filter_by(createdById=user_id)
.order_by(Book.title)
.limit(limit)
.offset(cursor))
books = builtin_list(map(from_sql, query.all()))
next_page = cursor + limit if len(books) == limit else None
return (books, next_page)
# [END list_by_user]
def read(id):
result = Book.query.get(id)
if not result:
return None
return from_sql(result)
def create(data):
book = Book(**data)
db.session.add(book)
db.session.commit()
return from_sql(book)
def update(data, id):
book = Book.query.get(id)
for k, v in data.items():
setattr(book, k, v)
db.session.commit()
return from_sql(book)
def delete(id):
Book.query.filter_by(id=id).delete()
db.session.commit()
def _create_database():
"""
If this script is run directly, create all the tables necessary to run the
application.
"""
app = Flask(__name__)
app.config.from_pyfile('../config.py')
init_app(app)
with app.app_context():
db.create_all()
print("All tables created")
if __name__ == '__main__':
_create_database()