डेटाबेस

te किसी विशेष डेटाबेस कार्यान्वयन से कड़ाई से जुड़ा नहीं है।

आप इसे GINO जैसे अतुल्यकालिक ORके साथ उपयोग कर सकते हैं, या नियमित गैर-अतुल्यकालिक समापन बिंदुओं का उपयोग कर सकते हैं और SQLAlchके साथ एकीकृत कर सकते हैं।

इस दस्तावेज़ में, हम प्रदर्शित करेंगे कि कैसे लक्षित किया जाए databases पैकेज एकीकरण, जो विभिन्न डेटाबेस ड्राइवरों की एक श्रृंखला के लिए SQLAlchकोर समर्थन प्रदान करता है।

यहाँ एक पूर्ण उदाहरण है, जिसमें तालिका परिभाषा, कॉन्फ़िगरेशन और शामिल है database.Database उदाहरण, और कई समापन बिंदु जो डेटाबेस के साथ बातचीत करते हैं।

**। Env **

DATABASE_URL = sqlite: //test.db

app.py

आयात करें

आयात डेटाबेस
आयात sqlalchआयात करें
te.से आयात
te.config से कॉन्फ़िग आयात करें
से te.आयात JSONआयात
स्टारलेट से। रूटिंग आयात मार्ग


# पर्यावरण चर या '.env' फ़ाइलों से कॉन्फ़िगरेशन।
कॉन्फ़िग = कॉन्फ़िग ('.env')
DATABASE_URL = कॉन्फ़िग ('DATABASE_URL')


# डेटाबेस तालिका परिभाषाएं।
मेटाडेटा = वर्गधर्मी। मेटाडेटा ()

नोट्स = वर्गमुखी। तालिका (
    "नोट्स",
    मेटाडेटा,
    Sqlalchemy। कॉलम ("आईडी", sqlalchemy। पूर्णांक, प्राथमिक_कुंजी = सच),
    Sqlalchemy। स्तंभ ("पाठ", sqlalchemy। स्ट्रिंग),
    Sqlalchemy। कॉलम ("पूरा", sqlalchemy। बूलियन),
)

डेटाबेस = डेटाबेस। डेटाबेस (DATABASE_URL)

@ lib.asyncप्रबंधक
Ac def जीवनकाल (app):
    प्रतीक्षा डेटाबेस कनेक्ट ()
    उपज
    प्रतीक्षा डेटाबेस .dis()

# मुख्य आवेदन कोड।
Ac def सूची _ नोट्स (अनुरोध):
    क्वेरी = notes.select ()
    परिणाम = प्रतीक्षा डेटाबेस। भ्रूण _ सभी (क्वेरी)
    सामग्री = [
        {
            "पाठ": परिणाम ["पाठ"],
            "पूर्ण": परिणाम ["पूर्ण"]
        }
        परिणामों में परिणामों के लिए
    ]
    JSONरिटर्न देता है (content)

Ac def add_note (request):
    डेटा = अनुरोध का इंतजार करें। json ()
    क्वेरी = no.() .मान (
       पाठ = डेटा ["पाठ"],
       पूर्ण = डेटा ["पूर्ण"]
    )
    प्रतीक्षारत डेटाबेस .(क्वेरी)
    JSONरिटर्न देता है ({)
        "पाठ": डेटा ["पाठ"],
        "पूर्ण": डेटा ["पूर्ण"]
    })

मार्ग = [
    मार्ग ("/ नोट्स", समापन बिंदु = सूची _ नोट्स, तरीके = ["GET"]),
    रूट ("/ नोट्स", एंडपॉइंट = add_note, तरीके = ["POST"]),
]

ऐप = स्टारलेट (
    मार्ग = मार्ग,
    जीवनकाल = जीवनकाल,
)

अंत में, आपको डेटा डेटाबेस और तालिका बनाने की आवश्यकता है। एलेम्बिक की सिफारिश की जाती है और हम इसे माइग्रेशन में संक्षेप में कवर करते हैं।

क्वेरी

SQLAlchकोर प्रश्नों का उपयोग करके क्वेरी की जा सकती है।

निम्नलिखित तरीके समर्थित हैं:

• rows = await database.fetch_all(query)
• row = await database.fetch_one(query)
• async for row in database.iterate(query)
• await database.execute(query)
• await database.execute_many(query)

लेनदेन; व्यापार; लेनदेन (इस शब्द के कई अर्थ हैं, और विशिष्ट अनुवाद संदर्भ पर निर्भर करता है)। यदि केवल दिए गए शब्द "लेनदेन" का शाब्दिक अनुवाद किया जाता है, तो इसका अनुवाद "लेनदेन; लेनदेन" के रूप में किया जा सकता है। विशिष्ट संदर्भ के अनुसार चुनने के लिए आपके लिए यहां कई सामान्य परिभाषाएँ हैं। शीर्षक की आवश्यकताओं के अनुसार, प्रत्यक्ष आउटपुट का अनुवाद इस प्रकार किया गया है: लेनदेन

---

डेटाबेस लेनदेन का उपयोग डेकोरेटर, संदर्भ प्रबंधक या निम्न-स्तरीय एपीआई के रूप में किया जा सकता है।

एंडपॉइंट पर डेकोरेटर का उपयोग करें:

@ डेटाबेस.ट्रांजेक्शन ()
Ac def populate_note (request):
    # यह डेटाबेस सम्मिलित एक लेनदेन के भीतर होता है।
    # इसे वापस रोल किया जाएगा `RuntimeError` हाँ।
    क्वेरी = no.() .मान (text="आप मुझे नहीं देखेंगे", पूरा = सच)
    प्रतीक्षारत डेटाबेस .(क्वेरी)
    उठाएँ ()

संदर्भ प्रबंधक का उपयोग करना:

Ac def populate_note (request):
    डेटाबेस लेनदेन () के साथ एसिंक:
        # यह डेटाबेस सम्मिलित एक लेनदेन के भीतर होता है।
        # इसे`RuntimeError` द्वारा वापस रोल किया जाएगा।
        क्वेरी = no.() .मान (text="आप मुझे नहीं देखेंगे", पूरा = सच)
        अनुरोध की प्रतीक्षा कर रहा है। डेटाबेस .(क्वेरी)
        उठाएँ ()

निम्न-स्तरीय एपीआई का उपयोग करना:

Ac def populate_note (request):
    लेन-देन = डेटाबेस का इंतजार करें ()
    कोशिश:
        # यह डेटाबेस सम्मिलित एक लेनदेन के भीतर होता है।
        # इसे वापस रोल किया जाएगा `RuntimeError` हाँ।
        क्वेरी = no.() .मान (text="आप मुझे नहीं देखेंगे", पूरा = सच)
        प्रतीक्षारत डेटाबेस .(क्वेरी)
        उठाएँ ()
    सिवाय:
        लेन-देन की प्रतीक्षा कर रहा है। ()
        उठाएँ
    अन्य:
        लेन-देन का इंतजार है। ()

परीक्षण अलगाव

डेटाबेस का उपयोग करने वाली सेवा के खिलाफ परीक्षण करते समय, कुछ चीजें हैं जिन्हें हम सुनिश्चित करना चाहते हैं। हमारी आवश्यकताएं होनी चाहिए:

• एक अलग डेटाबेस के साथ परीक्षण।
• हर बार जब कोई परीक्षण चलाया जाता है, तो एक नया परीक्षण डेटाबेस बनाया जाता है।
• सुनिश्चित करें कि प्रत्येक परीक्षण मामले के बीच डेटाबेस स्थिति एक दूसरे से अलग है।

इन आवश्यकताओं को पूरा करने के लिए, हमें इस तरह से अपने अनुप्रयोगों और परीक्षणों का निर्माण करने की आवश्यकता है:

te.से आयात
te.config से कॉन्फ़िग आयात करें
आयात डेटाबेस

कॉन्फ़िग = कॉन्फ़िग (".env")

परीक्षण = कॉन्फ़िग ('परीक्षण', कास्ट = बूल, डिफ़ॉल्ट = गलत)
DATABASE _ = कॉन्फ़िग ('DATABASE _ URL', कास्ट = डेटाबेस। डेटाबेस)
T_ DATABASE _ = DATABASE _ (डेटाबेस = 'test_' + DATABASE _ डेटाबेस)

# परीक्षण के दौरान 'force_rollback' का उपयोग करें, यह सुनिश्चित करने के लिए कि हम डेटाबेस परिवर्तनों को बनाए नहीं रखते हैं
# प्रत्येक परीक्षण मामले के बीच।
यदि परीक्षण:
    डेटाबेस = डेटाबेस। डेटाबेस (TEST _ DATABASE _ URL, बल _ रोलबैक = सच)
अन्य:
    डेटाबेस = डेटाबेस। डेटाबेस (DATABASE_URL)

अभ्यास रन के दौरान, हमें अभी भी स्थापित करने की आवश्यकता है TESTING , और परीक्षण डेटाबेस सेट करें। मान लीजिए कि हम उपयोग कर रहे हैं py.test यहाँ हमारा है conftest.py यह कैसा लग सकता है:

आयात पाइटेस्ट
te.config आयात पर्यावरण से
से te.आयात Testent
स्क्वैल्केमी आयात से create_engine
sqlalchemy_utils से डेटाबेस _ मौजूद है, _ डेटाबेस, ड्रॉप _ डेटाबेस बनाएं

# यह सेट करता है `os.environ`, लेकिन कुछ अतिरिक्त सुरक्षा प्रदान करता है।
# अगर हमने इसे एप्लिकेशन आयात के नीचे रखा, तो यह एक त्रुटि बढ़ाएगा
# हमें सूचित करना कि 'टेस्टिंग' पहले ही पर्यावरण से पढ़ा जा चुका था।
एनवायरन ['टेस्टिंग'] = 'ट्रू'

आयात ऐप


@ .(गुंजाइश = "सत्र", ऑटोहाउस = सच)
डिफ क्रिएट _ टेस्ट _ डेटाबेस ():
  """
  हर परीक्षण मामले पर एक साफ डेटाबेस बनाएं।
  सुरक्षा के लिए, हमें गर्भपात करना चाहिए यदि कोई डेटाबेस पहले से मौजूद है।

  हम उपयोग करते हैं `sqlalchemy_utils` लगातार कुछ सहायकों के लिए यहां पैक किया गया है
  डेटाबेस बनाना और छोड़ना।
  """
  उरल = str (app. TEST_DATABASE_URL)
  इंजन = create_engine (url)
  डेटाबेस _ मौजूद नहीं है (url), 'परीक्षण डेटाबेस पहले से मौजूद है। परीक्षण छोड़ना।'
  _ डेटाबेस बनाएं (url) # परीक्षण डेटाबेस बनाएं।
  मेटाडेटा। create_all (engine) # टेबल बनाएं।
  उपज # परीक्षण चलाएं।
  ड्रॉप _ डेटाबेस (url) # परीक्षण डेटाबेस ड्रॉप करें।


@ py.()
डिफ क्लाइंट ():
    """
    परीक्षण के मामलों में 'क्लाइंट' स्थिरता का उपयोग करते समय, हमें पूर्ण डेटाबेस मिलेगा
    परीक्षण मामलों के बीच भूमिका:

    डेफ टेस्ट_होमपेज (क्लाइंट):
        url = app.url_path_for ('होमपेज')
        प्रतिक्रिया = ग्राहक। (url)
        प्रतिक्रिया दें। status_code == 200
    """
    ग्राहक के रूप में Testent (ऐप) के साथ:
        उपज ग्राहक

प्रवासन

अपने डेटाबेस में वृद्धिशील परिवर्तनों को प्रबंधित करने के लिए, आपको लगभग निश्चित रूप से डेटाबेस माइग्रेशन का उपयोग करने की आवश्यकता होगी। इसके लिए, हम SQLAlchemy के लेखकों द्वारा लिखे गए एलेम्बिक की अत्यधिक अनुशंसा करते हैं।

$ पिप एलेम्बिक स्थापित करें
$ एलेम्बिक इनिट माइग्रेशन

अब आपको इसे सेट करने की आवश्यकता है ताकि एलेम्बिक कॉन्फ़िगर किए गए DATABASE_URL का संदर्भ दे और आपकी तालिका मेटाडेटा का उपयोग करे।

alembic.ini में निम्नलिखित पंक्तियों को हटाएं:

Sqlalurl = ड्राइवर: // उपयोगकर्ता: pass@localhost dbname

मौजूद है migrations/env.py में, आपको स्थापित करने की आवश्यकता है 'sqlalchemy.url' कॉन्फ़िगरेशन कुंजी और target_metadata चर। आपको कुछ इस तरह की आवश्यकता होगी:

# एलेम्बिक कॉन्फिग ऑब्जेक्ट।
संदर्भ = config

# हमारे DATABASE_URL और हमारी तालिका परिभाषाओं का उपयोग करने के लिए एलेम्बिक को कॉन्फ़िगर करें ...
आयात ऐप
Config.set_main_option ('sqlalchemyurl', str (app.DATABASE_URL))
लक्ष्य _ मेटाडेटा = .adata

...

फिर, ऊपर दिए गए हमारे नोट उदाहरण का उपयोग करते हुए, प्रारंभिक संशोधन बनाएं:

एलेम्बिक संशोधन -m "नोट्स टेबल बनाएं"

और नई फ़ाइल को आबाद करें (inside migrations/versions) आवश्यक निर्देशों के साथ:

डेफ अपग्रेड ():
    ऑप। _ तालिका बनाएँ (
      'नोट्स',
      Sqlalchemy। कॉलम ("आईडी", sqlalchemy। पूर्णांक, प्राथमिक_कुंजी = सच),
      Sqlalchemy। स्तंभ ("पाठ", sqlalchemy। स्ट्रिंग),
      Sqlalchemy। कॉलम ("पूरा", sqlalchemy। बूलियन),
    )

डिफ डाउनग्रेड ():
    ऑप। ड्रॉप _ टेबल ('नोट्स')

और अपना पहला माइग्रेशन चलाएं। हमारा नोट लेने वाला ऐप अब ऊपर और चल रहा है!

एलेम्बिक अपग्रेड हेड

परीक्षण के दौरान माइग्रेशन चलाएं

यह सुनिश्चित करना अच्छा अभ्यास है कि आपका परीक्षण सूट हर बार जब आप परीक्षण डेटाबेस बनाते हैं तो डेटाबेस माइग्रेशन चलाता है। यह आपके माइग्रेशन स्क्रिप्ट में किसी भी मुद्दे की पहचान करने में मदद करेगा और यह सुनिश्चित करने में मदद करेगा कि परीक्षण एक डेटाबेस के खिलाफ चलाए जाते हैं जो आपके लाइव डेटाबेस की स्थिति के अनुरूप है।

हम इसे थोड़ा समायोजित कर सकते हैं। create_test_database स्थिरता:

एलेम्बिक आयात कमांड से
alic.config से कॉन्फ़िग आयात करें
आयात ऐप

...

@ .(गुंजाइश = "सत्र", ऑटोहाउस = सच)
डिफ क्रिएट _ टेस्ट _ डेटाबेस ():
    उरल = str (app. DATABASE_URL)
    इंजन = create_engine (url)
    डेटाबेस _ मौजूद नहीं है (url), 'परीक्षण डेटाबेस पहले से मौजूद है। परीक्षण छोड़ना।'
    _ डेटाबेस बनाएं (url) # परीक्षण डेटाबेस बनाएं।
    कॉन्फ़िग = कॉन्फ़िग ("alembic.ini") # पलायन को चलाएं।
    कमांड.अपग्रेड (config, "head")
    उपज # परीक्षण चलाएं।
    ड्रॉप _ डेटाबेस (url) # परीक्षण डेटाबेस ड्रॉप करें।