-
Notifications
You must be signed in to change notification settings - Fork 36
Expand file tree
/
Copy pathsample_test.py
More file actions
86 lines (67 loc) · 2.87 KB
/
sample_test.py
File metadata and controls
86 lines (67 loc) · 2.87 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
import os
from fastapi import FastAPI
from sqlalchemy import ARRAY, BigInteger, Boolean, CHAR, Column, Date, DateTime, Float, Integer, \
JSON, Numeric, SmallInteger, String, Text, Time, UniqueConstraint, text
from sqlalchemy.dialects.postgresql import INTERVAL, JSONB, UUID
from sqlalchemy.orm import declarative_base, sessionmaker
from fastapi_quickcrud_codegen import crud_router_builder, CrudMethods
TEST_DATABASE_URL = os.environ.get('TEST_DATABASE_URL', 'postgresql://postgres:1234@127.0.0.1:5432/postgres')
app = FastAPI()
Base = declarative_base()
metadata = Base.metadata
from sqlalchemy import create_engine
engine = create_engine(TEST_DATABASE_URL, future=True, echo=True,
pool_use_lifo=True, pool_pre_ping=True, pool_recycle=7200)
async_session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_transaction_session():
try:
db = async_session()
yield db
finally:
db.close()
class UntitledTable256(Base):
primary_key_of_table = "primary_key"
unique_fields = ['primary_key', 'int4_value', 'float4_value']
__tablename__ = 'test_build_myself'
__table_args__ = (
UniqueConstraint('primary_key', 'int4_value', 'float4_value'),
)
primary_key = Column(Integer, primary_key=True, info={'alias_name': 'primary_key'}, autoincrement=True,
server_default="nextval('test_build_myself_id_seq'::regclass)")
bool_value = Column(Boolean, nullable=False, server_default=text("false"))
# bytea_value = Column(LargeBinary)
char_value = Column(CHAR(10))
date_value = Column(Date, server_default=text("now()"))
float4_value = Column(Float, nullable=False)
float8_value = Column(Float(53), nullable=False, server_default=text("10.10"))
int2_value = Column(SmallInteger, nullable=False)
int4_value = Column(Integer, nullable=True)
int8_value = Column(BigInteger, server_default=text("99"))
interval_value = Column(INTERVAL)
json_value = Column(JSON)
jsonb_value = Column(JSONB(astext_type=Text()))
numeric_value = Column(Numeric)
text_value = Column(Text)
time_value = Column(Time)
timestamp_value = Column(DateTime)
timestamptz_value = Column(DateTime(True))
timetz_value = Column(Time(True))
uuid_value = Column(UUID(as_uuid=True))
varchar_value = Column(String)
# xml_value = Column(NullType)
array_value = Column(ARRAY(Integer()))
array_str__value = Column(ARRAY(String()))
# box_valaue = Column(NullType)
crud_route_child2 = crud_router_builder(
db_model=UntitledTable256,
prefix="/blog_comment",
tags=["blog_comment"],
db_session=get_transaction_session
)
app = FastAPI()
[app.include_router(i) for i in [crud_route_child2]]
@app.get("/", tags=["child"])
async def root():
return {"message": "Hello World"}
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8002, debug=False)