다음을 통해 공유

빠른 시작: Microsoft Python Driver for SQL Server를 사용하여 Fabric의 SQL 데이터베이스에 연결

이 빠른 시작에서는 보고서를 빠르게 만드는 데 사용 Streamlit 하므로 사용자 피드백을 신속하게 수집하여 올바른 궤도에 있는지 확인할 수 있습니다. Python용 드라이버를 사용하여 mssql-pythonFabric의 SQL 데이터베이스에 연결하고 보고서에 로드된 데이터를 읽습니다.

드라이버에는 mssql-python Windows 컴퓨터에 대한 외부 종속성이 필요하지 않습니다. 드라이버는 단일 pip 설치로 필요한 모든 것을 설치하므로 업그레이드 및 테스트할 시간이 없는 다른 스크립트를 중단하지 않고도 새 스크립트에 최신 버전의 드라이버를 사용할 수 있습니다.

mssql-python 설명서 | mssql-python 소스 코드 | 패키지(PyPI) | Uv

필수 조건

프로젝트 만들기 및 코드 실행

새 프로젝트 만들기

  1. 개발 디렉터리에서 명령 프롬프트를 엽니다. 디렉터리가 없으면, python, scripts 등으로 새 디렉터리를 만드세요. OneDrive에는 폴더를 생성하지 마십시오. 동기화가 가상 환경 관리에 영향을 미칠 수 있습니다.

  2. 를 사용하여 새 프로젝트를 만듭니다 uv.

    uv init rapid-prototyping-qs
cd rapid-prototyping-qs

종속성 추가

동일한 디렉터리에 mssql-python, streamlit, 및 python-dotenv 패키지를 설치합니다.

uv add mssql-python python-dotenv streamlit

Visual Studio Code 시작

동일한 디렉터리에서 다음 명령을 실행합니다.

code .

pyproject.toml 업데이트

  1. pyproject.toml에는 프로젝트에 대한 메타데이터가 포함됩니다. 즐겨 찾는 편집기에서 파일을 엽니다.

  2. 설명을 더 자세히 설명하도록 업데이트합니다.

    description = "A quick example of rapid prototyping using the mssql-python driver and Streamlit."

  3. 파일을 저장 후 닫습니다.

main.py를 수정하십시오.

  1. 라는 main.py파일을 엽니다. 이 예제와 유사해야 합니다.

    def main():
 print("Hello from rapid-protyping-qs!")

 if __name__ == "__main__":
   main()

  2. 파일 맨 위에서 def main()줄 위에 다음 가져오기를 추가합니다.

    팁 (조언)

    Visual Studio Code에서 패키지를 해결하는 데 문제가 있는 경우 가상 환경을 사용하도록 인터프리터를 업데이트해야 합니다.

    from os import getenv
from dotenv import load_dotenv
from mssql_python import connect, Connection
import pandas as pd
import streamlit as st

  3. 가져오기와 줄 def main()사이에 다음 코드를 추가합니다.

    def page_load() -> None:
   st.set_page_config(
       page_title="View Data",
       page_icon=":bar_chart:",
       layout="wide",
       initial_sidebar_state="expanded"
   )

   st.title("AdventureWorksLT Customer Order History")

   SQL_QUERY = """SELECT c.* FROM [SalesLT].[Customer] c inner join SalesLT.SalesOrderHeader soh on c.CustomerId = soh.CustomerId;"""

   df = load_data(SQL_QUERY)

   event = st.dataframe(
       df,
       use_container_width=True,
       hide_index=True,
       on_select="rerun",
       selection_mode="single-row"
   )

   customer = event.selection.rows

   if len(customer) == 0:
       SQL_QUERY = """select soh.OrderDate, SUM(sod.OrderQty), SUM(sod.OrderQty * sod.UnitPrice) as spend, pc.Name as ProductCategory from SalesLT.SalesOrderDetail sod inner join SalesLT.SalesOrderHeader soh on sod.salesorderid = soh.salesorderid inner join SalesLT.Product p on sod.productid = p.productid inner join SalesLT.ProductCategory pc on p.ProductCategoryID = pc.ProductCategoryID GROUP BY soh.OrderDate, pc.Name ORDER BY soh.OrderDate, pc.Name;"""
       params = None
   else:
       customer_id = int(df.loc[customer, 'CustomerID'].values[0])
       SQL_QUERY = """select soh.OrderDate, SUM(sod.OrderQty), SUM(sod.OrderQty * sod.UnitPrice) as spend, pc.Name as ProductCategory from SalesLT.SalesOrderDetail sod inner join SalesLT.SalesOrderHeader soh on sod.salesorderid = soh.salesorderid inner join SalesLT.Product p on sod.productid = p.productid inner join SalesLT.ProductCategory pc on p.ProductCategoryID = pc.ProductCategoryID where soh.CustomerID = ? GROUP BY soh.OrderDate, pc.Name ORDER BY soh.OrderDate, pc.Name;"""
       params = (customer_id,)

   st.write("Here's a summary of spend by product category over time:")
   st.bar_chart(load_data(SQL_QUERY, params).set_index('ProductCategory')
                ['spend'], use_container_width=True)

   if len(customer) > 0:
       customer_id = int(df.loc[customer, 'CustomerID'].values[0])
       st.write(
           f"Displaying orders for Customer ID: {customer_id}")
       SQL_QUERY = """SELECT * FROM [SalesLT].[SalesOrderHeader] soh WHERE soh.CustomerID = ?;"""
       st.dataframe(load_data(SQL_QUERY, (customer_id,)), hide_index=True, use_container_width=True)
       SQL_QUERY = """SELECT sod.* FROM [SalesLT].[SalesOrderHeader] soh INNER JOIN SalesLT.SalesOrderDetail sod on soh.SalesOrderId = sod.SalesOrderId WHERE CustomerID = ?;"""
       st.dataframe(load_data(SQL_QUERY, (customer_id,)), hide_index=True, use_container_width=True)

  4. 가져오기 사이에 def page_load() -> None: 코드를 삽입합니다.

    _connection = None

def get_connection() -> Connection:
    global _connection
    if not _connection:
        load_dotenv()
        _connection = connect(getenv("SQL_CONNECTION_STRING"))
    return _connection

@st.cache_data
def load_data(SQL_QUERY, params=None) -> pd.DataFrame:
    data = pd.read_sql_query(SQL_QUERY, get_connection(), params=params)
    return data

  5. 이 코드를 찾습니다.

    def main():
    print("Hello from rapid-protyping-qs!")

  6. 이 코드로 대체합니다.

    def main() -> None:
    page_load()
    if _connection:
        _connection.close()

  7. 저장하고 닫습니다 main.py.

연결 문자열 저장

  1. .gitignore 파일을 열고 .env 파일에 대한 제외를 추가합니다. 파일은 이 예제와 유사해야 합니다. 완료되면 저장하고 닫아야 합니다.

    # Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env

  2. 현재 디렉터리에서 새 파일을 만듭니다 .env.

  3. .env 파일 내부에서 SQL_CONNECTION_STRING이라는 이름의 연결 문자열 항목을 추가합니다. 여기서 예제를 실제 연결 문자열 값으로 바꿉다.

    SQL_CONNECTION_STRING="Server=<server_name>;Database={<database_name>};Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"

uv run을 사용하여 스크립트 실행

팁 (조언)

macOS에서는 Microsoft Entra 인증을 위해 ActiveDirectoryInteractive 및 ActiveDirectoryDefault가 모두 작동합니다. ActiveDirectoryInteractive 는 스크립트를 실행할 때마다 로그인하라는 메시지를 표시합니다. 반복되는 로그인 프롬프트를 방지하려면 Azure CLI를 통해 한 번 로그인하고 ActiveDirectoryDefault를 사용하여 캐시된 자격 증명을 재사용하십시오.

  1. 이전의 터미널 창이나 동일한 디렉터리에 열려 있는 새 터미널 창에서 다음 명령을 실행합니다.

     uv run streamlit run main.py

  2. 보고서가 웹 브라우저의 새 탭에서 열립니다.

  3. 보고서의 작동 방식을 확인해 보세요. 변경한 내용이 있으면 브라우저 창의 오른쪽 위 모서리에 있는 다시 로드 옵션을 저장 main.py 하고 사용합니다.

  4. 프로토타입을 공유하려면 폴더를 .venv 제외한 모든 파일을 다른 컴퓨터에 복사합니다. 폴더는 .venv 첫 번째 실행으로 다시 만들어집니다.

다음 단계

더 많은 예제를 mssql-python 보려면 드라이버 GitHub 리포지토리를 방문하여 아이디어를 기여하거나 문제를 보고하세요.

GitHub의 mssql-python 드라이버

  • Last updated on